Hadoop实战：微博数据分析_基于hadoop的微博数据分析系统

项目需求

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

数据集

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

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

　　     　　　黄晓明       黄晓明        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

实验结果

以上就是博主为大家介绍的这一板块的主要内容，这都是博主自己的学习过程，希望能给大家带来一定的指导作用，有用的还望大家点个支持，如果对你没用也望包涵，有错误烦请指出。如有期待可关注博主以第一时间获取更新哦，谢谢！　

 版权声明：本文为博主原创文章，未经博主允许不得转载。