Hadoop实战：微博数据分析

发布时间：2024-12-11 00:44

项目需求

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

数据集

　　下面是部分数据，猛戳此链接下载完整数据集

　　数据格式：明星明星微博名称粉丝数关注数微博数

　　　　　黄晓明黄晓明 22616497 506 2011

　　　　　　张靓颖张靓颖 27878708 238 3846

　　　　　　羅志祥羅志祥 30763518 277 3843

　　　　　　劉嘉玲劉嘉玲 12631697 350 2057

　　　　　　李娜李娜 23309493 81 631

　　　　　　成龙成龙 22485765 5 758

　　　　　　　...

思路分析

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

程序

Weibo.java

package com.hadoop.WeiboCount;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

/*
数据格式：   明星    明星微博名称              粉丝数    关注数    微博数
           黄晓明   黄晓明                       22616497   506   2011
           张靓颖   张靓颖                       27878708   238   3846
           张成龙2012   张成龙2012               9813621       199   744
           羅志祥   羅志祥                       30763518   277   3843
           劉嘉玲   劉嘉玲                       12631697   350   2057
           吳君如大美女   吳君如大美女       18490338   190   412
           柯震東Kai   柯震東Kai               31337479   219   795
           李娜   李娜                       23309493   81   631
           徐小平   徐小平                       11659926   1929   13795
           唐嫣   唐嫣                       24301532   200   2391
           有斐君   有斐君                       8779383   577   4251
           孙燕姿   孙燕姿                       21213839   68   342
           成龙   成龙                       22485765   5   758
*/

public class WeiBo implements WritableComparable<Object> {
// 其实这里，跟TVPlayData和ScoreWritable一样的
// 注意： Hadoop通过Writable接口实现的序列化机制，不过没有提供比较功能，所以和java中的Comparable接口合并，提供一个接口WritableComparable。（自定义比较）
// Writable接口提供两个方法(write和readFields)。

   // 直接利用java的基本数据类型int，定义成员变量fan、followers、microblogs
   // 粉丝
    private int fan;
    // 关注
    private int followers;
    // 微博数
    private int microblogs;

// 问：这里我们自己编程时，是一定要创建一个带有参的构造方法，为什么还要显式的写出来一个带无参的构造方法呢？
// 答：构造器其实就是构造对象实例的方法，无参数的构造方法是默认的，但是如果你创造了一个带有参数的构造方法，那么无参的构造方法必须显式的写出来，否则会编译失败。

public WeiBo(){}; //java里的无参构造函数，是用来在创建对象时初始化对象
//在hadoop的每个自定义类型代码里，好比，现在的WeiBo，都必须要写无参构造函数。

            //问：为什么我们在编程的时候，需要创建一个带有参的构造方法？
   //答:就是能让赋值更灵活。构造一般就是初始化数值，你不想别人用你这个类的时候每次实例化都能用另一个构造动态初始化一些信息么(当然没有需要额外赋值就用默认的)。
    public WeiBo(int fan,int followers,int microblogs){
        this.fan = fan;
        this.followers = followers;
        this.microblogs = microblogs;
    }

            //问：其实set和get方法，这两个方法只是类中的setxxx和getxxx方法的总称，
   //那么，为什么在编程时，有set和set***两个，只有get***一个呢？
    public void set(int fan,int followers,int microblogs){
        this.fan = fan;
        this.followers = followers;
        this.microblogs = microblogs;
    }

        //   public float get(int fan,int followers,int microblogs){因为这是错误的，所以对于set可以分开，get只能是get***
//   return fan;
//   return followers;
//   return microblogs;
//}

       // 实现WritableComparable的readFields()方法，以便该数据能被序列化后完成网络传输或文件输入
//   对象不能传输的，需要转化成字节流！
//   将对象转换为字节流并写入到输出流out中是序列化，write 的过程（最好记!!!）
//   从输入流in中读取字节流反序列化为对象      是反序列化，readFields的过程（最好记!!!）

        @Override
    public void readFields(DataInput in) throws IOException {
       //拿代码来说的话，对象就是比如fan、followers。。。。
        fan = in.readInt(); //因为，我们这里的对象是Int类型，所以是readInt()
        followers = in.readInt();
        microblogs = in.readInt(); //注意:反序列化里，需要生成对象对吧，所以，是用到的是get对象
//       in.readByte()
//       in.readChar()
//       in.readDouble()
//       in.readLine()
//       in.readFloat()
//       in.readLong()
//       in.readShort()
    }

        // 实现WritableComparable的write()方法，以便该数据能被序列化后完成网络传输或文件输出
//   将对象转换为字节流并写入到输出流out中是序列化，write 的过程（最好记!!!）
//   从输入流in中读取字节流反序列化为对象      是反序列化，readFields的过程（最好记!!!）
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeInt(fan);
        out.writeInt(followers); //因为，我们这里的对象是Int类型，所以是writeInt()
        out.writeInt(microblogs); //注意:序列化里，需要对象对吧，所以，用到的是set那边的对象
//       out.writeByte()
//       out.writeChar()
//       out.writeDouble()
//       out.writeFloat()
//       out.writeLong()
//       out.writeShort()
//       out.writeUTF()
    }

         @Override
    public int compareTo(Object o) { //java里的比较，Java String.compareTo()
       // TODO Auto-generated method stub
        return 0;
    }

   //   Hadoop中定义了两个序列化相关的接口：Writable 接口和 Comparable 接口，这两个接口可以合并成一个接口 WritableComparable。
//   Writable 接口中定义了两个方法，分别为write(DataOutput out)和readFields(DataInput in)
//   所有实现了Comparable接口的对象都可以和自身相同类型的对象比较大小

   // 源码是
//   package java.lang;
//   import java.util.*;
//   public interface Comparable {
//       /**
//       * 将this对象和对象o进行比较，约定：返回负数为小于，零为大于，整数为大于
//       */
//       public int compareTo(T o);
//   }

    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;
    }
}

WeiboCount.java

package com.hadoop.WeiboCount;

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;

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){ //增强型for循环，意思是把Values的值传给Text value
                // 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://centpy:9000/weibo/weibo.txt", "hdfs://centpy:9000/weibo/out/" };

//      String[] args0 = { "./data/weibo/weibo.txt", "./out/weibo/" };
        int ec = ToolRunner.run(new Configuration(), new WeiboCount(), args0);
        System.exit(ec);
    }
}

WeiboInputFormat.java

package com.hadoop.WeiboCount;

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.JobContext;
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;

//其实这个程序，就是在实现InputFormat接口，TVPlayInputFormat是InputFormat接口的实现类
//比如 WeiboInputFormat extends FileInputFormat implements InputFormat。

//问：自定义输入格式 WeiboInputFormat 类，首先继承 FileInputFormat，然后分别重写 isSplitable() 方法和 createRecordReader() 方法。

//线路是： boolean isSplitable() -> RecordReader<Text,WeiBo> createRecordReader() -> WeiboRecordReader extends RecordReader<Text, WeiBo >
public class WeiboInputFormat extends FileInputFormat<Text,WeiBo>{

   @Override
   protected boolean isSplitable(JobContext context, Path filename) {//这是InputFormat的isSplitable方法
           //isSplitable方法就是是否要切分文件，这个方法显示如果是压缩文件就不切分，非压缩文件就切分。
//       如果不允许分割，则isSplitable==false，则将第一个block、文件目录、开始位置为0，长度为整个文件的长度封装到一个InputSplit，加入splits中
//       如果文件长度不为0且支持分割，则isSplitable==true,获取block大小，默认是64MB
       return false;    //整个文件封装到一个InputSplit
       //要么就是return true;       //切分64MB大小的一块一块，再封装到InputSplit
   }

   @Override
   public RecordReader<Text, WeiBo> createRecordReader(InputSplit arg0, TaskAttemptContext arg1) throws IOException, InterruptedException {
//       RecordReader<k1, v1>是返回类型,返回的RecordReader对象的封装
//       createRecordReader是方法,在这里是，WeiboInputFormat.createRecordReader。WeiboInputFormat是InputFormat类的实例
//       InputSplit input和TaskAttemptContext context是传入参数

//       isSplitable(),如果是压缩文件就不切分，整个文件封装到一个InputSplit
//       isSplitable(),如果是非压缩文件就切，切分64MB大小的一块一块，再封装到InputSplit

       //这里默认是系统实现的的RecordReader，按行读取，下面我们自定义这个类WeiboRecordReader。
       //类似与Excel、WeiBo、TVPlayData代码写法

   // 自定义WeiboRecordReader类，按行读取
        return new WeiboRecordReader(); //新建一个ScoreRecordReader实例，所有才有了上面RecordReader<Text,ScoreWritable>，所以才如下ScoreRecordReader，写我们自己的
   }

   public class WeiboRecordReader extends RecordReader<Text, WeiBo>{
       //LineReader      in是1，行号。
           //Text line;      俞灏明   俞灏明   10591367   206   558，每行的相关记录
      public LineReader in; //行读取器
      public Text line;//每行数据类型
       public Text lineKey;//自定义key类型，即k1
       public WeiBo lineValue;//自定义value类型，即v1

          @Override
          public void initialize(InputSplit input, TaskAttemptContext context) throws IOException, InterruptedException { //初始化，都是模板

              FileSplit split = (FileSplit)input; // 获取split

              Configuration job = context.getConfiguration(); // 获取配置

              Path file = split.getPath(); // 分片路径

              FileSystem fs = file.getFileSystem(job);

              FSDataInputStream filein = fs.open(file);   // 打开文件

              in=new LineReader(filein,job); //输入流in
           line=new Text();//每行数据类型
           lineKey=new Text();//自定义key类型，即k1。//新建一个Text实例作为自定义格式输入的key
           lineValue = new WeiBo();//自定义value类型，即v1。//新建一个TVPlayData实例作为自定义格式输入的value
          }

       //此方法读取每行数据，完成自定义的key和value
          @Override
          public boolean nextKeyValue() throws IOException, InterruptedException { //这里面，才是篡改的重点
              // 一行数据
              Text line = new Text();

              int linesize = in.readLine(line); //line是每行数据，我们这里用到的是in.readLine(str)这个构造函数，默认读完读到文件末尾。其实这里有三种。

//            是SplitLineReader.readLine -> SplitLineReader extends   LineReader -> org.apache.hadoop.util.LineReader

//            in.readLine(str)//这个构造方法执行时，会首先将value原来的值清空。默认读完读到文件末尾
//            in.readLine(str, maxLineLength)//只读到maxLineLength行
//            in.readLine(str, maxLineLength, maxBytesToConsume)//这个构造方法来实现不清空，前面读取的行的值

              if(linesize == 0)
                  return false;

              // 通过分隔符'\t'，将每行的数据解析成数组 pieces
              String[] pieces = line.toString().split("\t");//因为，我们这里是。默认读完读到文件末尾。line是Text类型。pieces是String[]，即String数组。

              if(pieces.length != 5){
                  throw new IOException("Invalid record received");
              }

              int a,b,c;
              try{

                  a = Integer.parseInt(pieces[2].trim()); //粉丝数,//将String类型，如pieces[2]转换成，float类型，给a

                  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]); //完成自定义key数据
              lineValue.set(a, b, c); //完成自定义value数据

//            或者写
//            lineValue.set(Integer.parseInt(pieces[2].trim()),Integer.parseInt(pieces[3].trim()),Integer.parseInt(pieces[4].trim()));

              return true;
          }

          @Override
          public void close() throws IOException { //关闭输入流
              if(in != null){
                  in.close();
              }
          }

          @Override
          public Text getCurrentKey() throws IOException, InterruptedException { //获取当前的key,即CurrentKey
              return lineKey; //返回类型是Text,即Text lineKey
          }

          @Override
          public WeiBo getCurrentValue() throws IOException, InterruptedException { //获取当前的Value，即CurrentValue
              return lineValue; //返回类型是WeiBo,即WeiBo lineValue
          }

          @Override
          public float getProgress() throws IOException, InterruptedException { //获取进程，即Progress
              return 0; //返回类型是float,即float 0
          }

      }
}

实验结果