利用 MapReduce分析明星微博数据实战-低调大师

利用 MapReduce分析明星微博数据实战

2017-07-31 485

互联网时代的到来，使得名人的形象变得更加鲜活，也拉近了明星和粉丝之间的距离。歌星、影星、体育明星、作家等名人通过互联网能够轻易实现和粉丝的互动，赚钱也变得前所未有的简单。同时，互联网的飞速发展本身也造就了一批互联网明星，这些人借助新的手段，最大程度发挥了粉丝经济的能量和作用，在互联网时代赚得盆满钵满。

正是基于这样一个大背景，今天我们做一个分析明星微博数据的小项目。

1、项目需求

自定义输入格式，将明星微博数据排序后按粉丝数关注数微博数分别输出到不同文件中。

2、数据集

明星明星微博名称粉丝数关注数微博数

俞灏明俞灏明 10591367 206 558

李敏镐李敏镐 22898071 11 268

林心如林心如 57488649 214 5940

黄晓明黄晓明 22616497 506 2011

张靓颖张靓颖 27878708 238 3846

李娜李娜 23309493 81 631

徐小平徐小平 11659926 1929 13795

唐嫣唐嫣 24301532 200 2391

有斐君有斐君 8779383 577 4251

3、分析

自定义InputFormat读取明星微博数据，通过自定义getSortedHashtableByValue方法分别对明星的fan、followers、microblogs数据进行排序，然后利用MultipleOutputs输出不同项到不同的文件中

4、实现

1）、定义WeiBo实体类，实现WritableComparable接口

package com.buaa;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.io.WritableComparable;
/**
* @ProjectName MicroblogStar
* @PackageName com.buaa
* @ClassName WeiBo
* @Description TODO
* @Author 刘吉超
* @Date 2016-05-07 14:54:29
*/
public class WeiBo implements WritableComparable<Object> {
// 粉丝
private int fan;
// 关注
private int followers;
// 微博数
private int microblogs;
public WeiBo(){};
public WeiBo(int fan,int followers,int microblogs){
this.fan = fan;
this.followers = followers;
this.microblogs = microblogs;
}
public void set(int fan,int followers,int microblogs){
this.fan = fan;
this.followers = followers;
this.microblogs = microblogs;
}
// 实现WritableComparable的readFields()方法，以便该数据能被序列化后完成网络传输或文件输入
@Override
public void readFields(DataInput in) throws IOException {
fan = in.readInt();
followers = in.readInt();
microblogs = in.readInt();
}
// 实现WritableComparable的write()方法，以便该数据能被序列化后完成网络传输或文件输出
@Override
public void write(DataOutput out) throws IOException {
out.writeInt(fan);
out.writeInt(followers);
out.writeInt(microblogs);
}
@Override
public int compareTo(Object o) {
// TODO Auto-generated method stub
return 0;
}
public int getFan() {
return fan;
}
public void setFan(int fan) {
this.fan = fan;
}
public int getFollowers() {
return followers;
}
public void setFollowers(int followers) {
this.followers = followers;
}
public int getMicroblogs() {
return microblogs;
}
public void setMicroblogs(int microblogs) {
this.microblogs = microblogs;
}
}

2）、自定义WeiboInputFormat，继承FileInputFormat抽象类

package com.buaa;
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;
/**
* @ProjectName MicroblogStar
* @PackageName com.buaa
* @ClassName WeiboInputFormat
* @Description TODO
* @Author 刘吉超
* @Date 2016-05-07 10:23:28
*/
public class WeiboInputFormat extends FileInputFormat<Text,WeiBo>{
@Override
public RecordReader<Text, WeiBo> createRecordReader(InputSplit arg0, TaskAttemptContext arg1) throws IOException, InterruptedException {
// 自定义WeiboRecordReader类，按行读取
return new WeiboRecordReader();
}
public class WeiboRecordReader extends RecordReader<Text, WeiBo>{
public LineReader in;
// 声明key类型
public Text lineKey = new Text();
// 声明 value类型
public WeiBo lineValue = new WeiBo();
@Override
public void initialize(InputSplit input, TaskAttemptContext context) throws IOException, InterruptedException {
// 获取split
FileSplit split = (FileSplit)input;
// 获取配置
Configuration job = context.getConfiguration();
// 分片路径
Path file = split.getPath();
FileSystem fs = file.getFileSystem(job);
// 打开文件
FSDataInputStream filein = fs.open(file);
in = new LineReader(filein,job);
}
@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
// 一行数据
Text line = new Text();
int linesize = in.readLine(line);
if(linesize == 0)
return false;
// 通过分隔符'\t'，将每行的数据解析成数组
String[] pieces = line.toString().split("\t");
if(pieces.length != 5){
throw new IOException("Invalid record received");
}
int a,b,c;
try{
// 粉丝
a = Integer.parseInt(pieces[2].trim());
// 关注
b = Integer.parseInt(pieces[3].trim());
// 微博数
c = Integer.parseInt(pieces[4].trim());
}catch(NumberFormatException nfe){
throw new IOException("Error parsing floating poing value in record");
}
//自定义key和value值
lineKey.set(pieces[0]);
lineValue.set(a, b, c);
return true;
}
@Override
public void close() throws IOException {
if(in != null){
in.close();
}
}
@Override
public Text getCurrentKey() throws IOException, InterruptedException {
return lineKey;
}
@Override
public WeiBo getCurrentValue() throws IOException, InterruptedException {
return lineValue;
}
@Override
public float getProgress() throws IOException, InterruptedException {
return 0;
}
}
}

3）、编写mr程序

package com.buaa;
import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
/**
* @ProjectName MicroblogStar
* @PackageName com.buaa
* @ClassName WeiboCount
* @Description TODO
* @Author 刘吉超
* @Date 2016-05-07 09:07:36
*/
public class WeiboCount extends Configured implements Tool {
// tab分隔符
private static String TAB_SEPARATOR = "\t";
// 粉丝
private static String FAN = "fan";
// 关注
private static String FOLLOWERS = "followers";
// 微博数
private static String MICROBLOGS = "microblogs";
public static class WeiBoMapper extends Mapper<Text, WeiBo, Text, Text> {
@Override
protected void map(Text key, WeiBo value, Context context) throws IOException, InterruptedException {
// 粉丝
context.write(new Text(FAN), new Text(key.toString() + TAB_SEPARATOR + value.getFan()));
// 关注
context.write(new Text(FOLLOWERS), new Text(key.toString() + TAB_SEPARATOR + value.getFollowers()));
// 微博数
context.write(new Text(MICROBLOGS), new Text(key.toString() + TAB_SEPARATOR + value.getMicroblogs()));
}
}
public static class WeiBoReducer extends Reducer<Text, Text, Text, IntWritable> {
private MultipleOutputs<Text, IntWritable> mos;
protected void setup(Context context) throws IOException, InterruptedException {
mos = new MultipleOutputs<Text, IntWritable>(context);
}
protected void reduce(Text Key, Iterable<Text> Values,Context context) throws IOException, InterruptedException {
Map<String,Integer> map = new HashMap< String,Integer>();
for(Text value : Values){
// value = 名称 + (粉丝数或关注数或微博数)
String[] records = value.toString().split(TAB_SEPARATOR);
map.put(records[0], Integer.parseInt(records[1].toString()));
}
// 对Map内的数据进行排序
Map.Entry<String, Integer>[] entries = getSortedHashtableByValue(map);
for(int i = 0; i < entries.length;i++){
mos.write(Key.toString(),entries[i].getKey(), entries[i].getValue());
}
}
protected void cleanup(Context context) throws IOException, InterruptedException {
mos.close();
}
}
@SuppressWarnings("deprecation")
@Override
public int run(String[] args) throws Exception {
// 配置文件对象
Configuration conf = new Configuration();
// 判断路径是否存在，如果存在，则删除
Path mypath = new Path(args[1]);
FileSystem hdfs = mypath.getFileSystem(conf);
if (hdfs.isDirectory(mypath)) {
hdfs.delete(mypath, true);
}
// 构造任务
Job job = new Job(conf, "weibo");
// 主类
job.setJarByClass(WeiboCount.class);
// Mapper
job.setMapperClass(WeiBoMapper.class);
// Mapper key输出类型
job.setMapOutputKeyClass(Text.class);
// Mapper value输出类型
job.setMapOutputValueClass(Text.class);
// Reducer
job.setReducerClass(WeiBoReducer.class);
// Reducer key输出类型
job.setOutputKeyClass(Text.class);
// Reducer value输出类型
job.setOutputValueClass(IntWritable.class);
// 输入路径
FileInputFormat.addInputPath(job, new Path(args[0]));
// 输出路径
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 自定义输入格式
job.setInputFormatClass(WeiboInputFormat.class) ;
//自定义文件输出类别
MultipleOutputs.addNamedOutput(job, FAN, TextOutputFormat.class, Text.class, IntWritable.class);
MultipleOutputs.addNamedOutput(job, FOLLOWERS, TextOutputFormat.class, Text.class, IntWritable.class);
MultipleOutputs.addNamedOutput(job, MICROBLOGS, TextOutputFormat.class, Text.class, IntWritable.class);
// 去掉job设置outputFormatClass，改为通过LazyOutputFormat设置
LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
//提交任务
return job.waitForCompletion(true)?0:1;
}
// 对Map内的数据进行排序（只适合小数据量）
@SuppressWarnings("unchecked")
public static Entry<String, Integer>[] getSortedHashtableByValue(Map<String, Integer> h) {
Entry<String, Integer>[] entries = (Entry<String, Integer>[]) h.entrySet().toArray(new Entry[0]);
// 排序
Arrays.sort(entries, new Comparator<Entry<String, Integer>>() {
public int compare(Entry<String, Integer> entry1, Entry<String, Integer> entry2) {
return entry2.getValue().compareTo(entry1.getValue());
}
});
return entries;
}
public static void main(String[] args) throws Exception {
String[] args0 = {
"hdfs://ljc:9000/buaa/microblog/weibo.txt",
"hdfs://ljc:9000/buaa/microblog/out/"
};
int ec = ToolRunner.run(new Configuration(), new WeiboCount(), args0);
System.exit(ec);
}
}

5、运行结果

本文作者：刘超ljc

来源：51CTO

微信关注我们

原文链接：https://yq.aliyun.com/articles/196082

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

2016 软件开发的七大趋势：容器技术将统治世界

岁月如割，2016距离我们只有几个“双xx”购物节之遥，对于“开发狗”来说，如死亡丛林飞跃般的练级之旅注定颠沛流离，命运多舛。是时候抬头看路了，下面我们为大家大胆假设一下2016的七大开发趋势。一、容器技术(Container)将统治世界毫无疑问当红炸子鸡Docker将度过叛逆青春期，变得更加成熟稳重，给人以安全感，更容易管理。二、Java将加速衰落虽然还有很多招聘说明中要求Java技能，Java也依然是目前主流的“吃饭”语言，但实际上这些企业热心的是Spark、Node.js或MongoDB，他们招你进去是为了让你能上手Spark，这意味着你真正要掌握的是Python和Scala。三、留神戴尔EMC的灾难性组合从IT历史规律来看，戴尔并购EMC是徒劳和不得人心的，恐龙和恐龙合体的场面通常很壮观，但最终的产物注定依然只能是一只反射弧超长的恐龙，开发者们，离恐龙远点。四、NoSQL数据库将站稳脚跟 NoSQL这个名字起得很失败，被各路砖家拍得也很惨。但是这挡不住2016年越来越多的大型传统企业开始使用NoSQL数据库。五、一切都实时化不仅仅是数据分析的实时化(快数据)，...

2017-08-01

437

在大数据圈你不知道的15个新技术

对于大数据来说，业界这几周算是比较忙碌的。因为很多初创公司和一些老牌的公司都推出了数据分析和数据管理产品，以及更新了现有产品，提供更丰富的功能与性能。虽然这些技术都还只是蓝图规划，但是一些常见的主题还是对其贯穿始终：为用户提供简单的获得数据的访问方式，更好的管理大规模数据以及预先分析的功能，例如Spark、HAWQ 和 Geode等新兴的大数据技术来实现更多的功能。下面一起来看看吸引眼球的十五项大数据公告。虽然罗列了很多，但还不是全部内容，只是最近在加利福尼亚州圣何塞市Strata + Hadoop World大会上亮相的部分技术。 Altiscale Insight Cloud Altiscale作为大数据服务的提供者，最近推出了Altiscale Insight Cloud产品，是一个自主服务型的数据分析服务产品，能够帮助业务分析人员使用BI工具等类似于Tableau和Excel的产品，实现快速的数据湖查询，而不需要IT部门投入过多的精力参与。 Altiscal Insight Cloud可以驱动SQL查询、动态可视化、实时仪表板和其他报告及分析能力。它不再因为聚合数据而需要一...

2017-08-01

374

发表评论

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Eclipse

Eclipse 是一个开放源代码的、基于Java的可扩展开发平台。就其本身而言，它只是一个框架和一组服务，用于通过插件组件构建开发环境。幸运的是，Eclipse 附带了一个标准的插件集，包括Java开发工具（Java Development Kit，JDK）。

JDK

JDK是 Java 语言的软件开发工具包，主要用于移动设备、嵌入式设备上的java应用程序。JDK是整个java开发的核心，它包含了JAVA的运行环境（JVM+Java系统类库）和JAVA工具。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。