鲨鱼网
利用 MapReduce分析明星微博数据实战
创始人
2025-01-29 00:50:26
0次

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

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

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接口

  1. package com.buaa; 
  2.  
  3. import java.io.DataInput; 
  4. import java.io.DataOutput; 
  5. import java.io.IOException; 
  6.  
  7. import org.apache.hadoop.io.WritableComparable; 
  8.  
  9. /**  
  10. * @ProjectName MicroblogStar 
  11. * @PackageName com.buaa 
  12. * @ClassName WeiBo 
  13. * @Description TODO 
  14. * @Author 刘吉超 
  15. * @Date 2016-05-07 14:54:29 
  16. */ 
  17. public class WeiBo implements WritableComparable { 
  18.     // 粉丝 
  19.     private int fan; 
  20.     // 关注 
  21.     private int followers; 
  22.     // 微博数 
  23.     private int microblogs; 
  24.      
  25.     public WeiBo(){}; 
  26.      
  27.     public WeiBo(int fan,int followers,int microblogs){ 
  28.         this.fan = fan; 
  29.         this.followers = followers; 
  30.         this.microblogs = microblogs; 
  31.     } 
  32.      
  33.     public void set(int fan,int followers,int microblogs){ 
  34.         this.fan = fan; 
  35.         this.followers = followers; 
  36.         this.microblogs = microblogs; 
  37.     } 
  38.      
  39.     // 实现WritableComparable的readFields()方法，以便该数据能被序列化后完成网络传输或文件输入 
  40.     @Override 
  41.     public void readFields(DataInput in) throws IOException { 
  42.         fan  = in.readInt(); 
  43.         followers = in.readInt(); 
  44.         microblogs = in.readInt(); 
  45.     } 
  46.      
  47.     // 实现WritableComparable的write()方法，以便该数据能被序列化后完成网络传输或文件输出  
  48.     @Override 
  49.     public void write(DataOutput out) throws IOException { 
  50.         out.writeInt(fan); 
  51.         out.writeInt(followers); 
  52.         out.writeInt(microblogs); 
  53.     } 
  54.      
  55.     @Override 
  56.     public int compareTo(Object o) { 
  57.         // TODO Auto-generated method stub 
  58.         return 0; 
  59.     } 
  60.  
  61.     public int getFan() { 
  62.         return fan; 
  63.     } 
  64.  
  65.     public void setFan(int fan) { 
  66.         this.fan = fan; 
  67.     } 
  68.  
  69.     public int getFollowers() { 
  70.         return followers; 
  71.     } 
  72.  
  73.     public void setFollowers(int followers) { 
  74.         this.followers = followers; 
  75.     } 
  76.  
  77.     public int getMicroblogs() { 
  78.         return microblogs; 
  79.     } 
  80.  
  81.     public void setMicroblogs(int microblogs) { 
  82.         this.microblogs = microblogs; 
  83.     } 
  84. } 
    85. 

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

    1. package com.buaa; 
    2.  
    3. import java.io.IOException; 
    4.  
    5. import org.apache.hadoop.conf.Configuration; 
    6. import org.apache.hadoop.fs.FSDataInputStream; 
    7. import org.apache.hadoop.fs.FileSystem; 
    8. import org.apache.hadoop.fs.Path; 
    9. import org.apache.hadoop.io.Text; 
    10. import org.apache.hadoop.mapreduce.InputSplit; 
    11. import org.apache.hadoop.mapreduce.RecordReader; 
    12. import org.apache.hadoop.mapreduce.TaskAttemptContext; 
    13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
    14. import org.apache.hadoop.mapreduce.lib.input.FileSplit; 
    15. import org.apache.hadoop.util.LineReader; 
    16.  
    17. /**  
    18. * @ProjectName MicroblogStar 
    19. * @PackageName com.buaa 
    20. * @ClassName WeiboInputFormat 
    21. * @Description TODO 
    22. * @Author 刘吉超 
    23. * @Date 2016-05-07 10:23:28 
    24. */ 
    25. public class WeiboInputFormat extends FileInputFormat{ 
    26.  
    27.      @Override 
    28.      public RecordReader createRecordReader(InputSplit arg0, TaskAttemptContext arg1) throws IOException, InterruptedException { 
    29.           // 自定义WeiboRecordReader类，按行读取 
    30.           return new WeiboRecordReader(); 
    31.      } 
    32.  
    33.      public class WeiboRecordReader extends RecordReader{ 
    34.             public LineReader in;  
    35.             // 声明key类型 
    36.             public Text lineKey = new Text(); 
    37.             // 声明 value类型 
    38.             public WeiBo lineValue = new WeiBo(); 
    39.              
    40.             @Override 
    41.             public void initialize(InputSplit input, TaskAttemptContext context) throws IOException, InterruptedException { 
    42.                 // 获取split 
    43.                 FileSplit split = (FileSplit)input; 
    44.                 // 获取配置  
    45.                 Configuration job = context.getConfiguration(); 
    46.                 // 分片路径  
    47.                 Path file = split.getPath(); 
    48.                  
    49.                 FileSystem fs = file.getFileSystem(job);  
    50.                 // 打开文件    
    51.                 FSDataInputStream filein = fs.open(file); 
    52.                  
    53.                 in = new LineReader(filein,job);  
    54.             } 
    55.  
    56.             @Override 
    57.             public boolean nextKeyValue() throws IOException, InterruptedException { 
    58.                 // 一行数据 
    59.                 Text line = new Text(); 
    60.                  
    61.                 int linesize = in.readLine(line); 
    62.                  
    63.                 if(linesize == 0)  
    64.                     return false;  
    65.                  
    66.                 // 通过分隔符'\t'，将每行的数据解析成数组 
    67.                 String[] pieces = line.toString().split("\t"); 
    68.                  
    69.                 if(pieces.length != 5){   
    70.                     throw new IOException("Invalid record received");   
    71.                 }  
    72.                  
    73.                 int a,b,c; 
    74.                 try{   
    75.                     // 粉丝   
    76.                     a = Integer.parseInt(pieces[2].trim()); 
    77.                     // 关注 
    78.                     b = Integer.parseInt(pieces[3].trim()); 
    79.                     // 微博数 
    80.                     c = Integer.parseInt(pieces[4].trim()); 
    81.                 }catch(NumberFormatException nfe){   
    82.                     throw new IOException("Error parsing floating poing value in record");   
    83.                 } 
    84.                  
    85.                 //自定义key和value值 
    86.                 lineKey.set(pieces[0]);   
    87.                 lineValue.set(a, b, c); 
    88.                  
    89.                 return true; 
    90.             } 
    91.              
    92.             @Override 
    93.             public void close() throws IOException { 
    94.                 if(in != null){ 
    95.                     in.close(); 
    96.                 } 
    97.             } 
    98.  
    99.             @Override 
    100.             public Text getCurrentKey() throws IOException, InterruptedException { 
    101.                 return lineKey; 
    102.             } 
    103.  
    104.             @Override 
    105.             public WeiBo getCurrentValue() throws IOException, InterruptedException { 
    106.                 return lineValue; 
    107.             } 
    108.  
    109.             @Override 
    110.             public float getProgress() throws IOException, InterruptedException { 
    111.                 return 0; 
    112.             } 
    113.              
    114.         } 
    115. } 
    

    3）、编写mr程序
    

    1. package com.buaa; 
    2.  
    3. import java.io.IOException; 
    4. import java.util.Arrays; 
    5. import java.util.Comparator; 
    6. import java.util.HashMap; 
    7. import java.util.Map; 
    8. import java.util.Map.Entry; 
    9.  
    10. import org.apache.hadoop.conf.Configuration; 
    11. import org.apache.hadoop.conf.Configured; 
    12. import org.apache.hadoop.fs.FileSystem; 
    13. import org.apache.hadoop.fs.Path; 
    14. import org.apache.hadoop.io.IntWritable; 
    15. import org.apache.hadoop.io.Text; 
    16. import org.apache.hadoop.mapreduce.Job; 
    17. import org.apache.hadoop.mapreduce.Mapper; 
    18. import org.apache.hadoop.mapreduce.Reducer; 
    19. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
    20. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
    21. import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat; 
    22. import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs; 
    23. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; 
    24. import org.apache.hadoop.util.Tool; 
    25. import org.apache.hadoop.util.ToolRunner; 
    26.  
    27. /**  
    28. * @ProjectName MicroblogStar 
    29. * @PackageName com.buaa 
    30. * @ClassName WeiboCount 
    31. * @Description TODO 
    32. * @Author 刘吉超 
    33. * @Date 2016-05-07 09:07:36 
    34. */ 
    35. public class WeiboCount extends Configured implements Tool { 
    36.     // tab分隔符 
    37.     private static String TAB_SEPARATOR = "\t"; 
    38.     // 粉丝 
    39.     private static String FAN = "fan"; 
    40.     // 关注 
    41.     private static String FOLLOWERS = "followers"; 
    42.     // 微博数 
    43.     private static String MICROBLOGS = "microblogs"; 
    44.      
    45.     public static class WeiBoMapper extends Mapper { 
    46.         @Override 
    47.         protected void map(Text key, WeiBo value, Context context) throws IOException, InterruptedException { 
    48.             // 粉丝 
    49.             context.write(new Text(FAN), new Text(key.toString() + TAB_SEPARATOR + value.getFan())); 
    50.             // 关注 
    51.             context.write(new Text(FOLLOWERS), new Text(key.toString() + TAB_SEPARATOR + value.getFollowers())); 
    52.             // 微博数 
    53.             context.write(new Text(MICROBLOGS), new Text(key.toString() + TAB_SEPARATOR + value.getMicroblogs())); 
    54.         } 
    55.     } 
    56.      
    57.     public static class WeiBoReducer extends Reducer { 
    58.         private MultipleOutputs mos; 
    59.  
    60.         protected void setup(Context context) throws IOException, InterruptedException { 
    61.             mos = new MultipleOutputs(context); 
    62.         } 
    63.  
    64.         protected void reduce(Text Key, Iterable Values,Context context) throws IOException, InterruptedException { 
    65.             Map map = new HashMap< String,Integer>(); 
    66.              
    67.             for(Text value : Values){ 
    68.                 // value = 名称 + (粉丝数 或 关注数 或 微博数) 
    69.                 String[] records = value.toString().split(TAB_SEPARATOR); 
    70.                 map.put(records[0], Integer.parseInt(records[1].toString())); 
    71.             } 
    72.              
    73.             // 对Map内的数据进行排序 
    74.             Map.Entry[] entries = getSortedHashtableByValue(map); 
    75.              
    76.             for(int i = 0; i < entries.length;i++){ 
    77.                 mos.write(Key.toString(),entries[i].getKey(), entries[i].getValue()); 
    78.             }                
    79.         } 
    80.  
    81.         protected void cleanup(Context context) throws IOException, InterruptedException { 
    82.             mos.close(); 
    83.         } 
    84.     } 
    85.      
    86.     @SuppressWarnings("deprecation") 
    87.     @Override 
    88.     public int run(String[] args) throws Exception { 
    89.         // 配置文件对象 
    90.         Configuration conf = new Configuration(); 
    91.          
    92.         // 判断路径是否存在，如果存在，则删除 
    93.         Path mypath = new Path(args[1]); 
    94.         FileSystem hdfs = mypath.getFileSystem(conf); 
    95.         if (hdfs.isDirectory(mypath)) { 
    96.             hdfs.delete(mypath, true); 
    97.         } 
    98.          
    99.         // 构造任务 
    100.         Job job = new Job(conf, "weibo"); 
    101.         // 主类 
    102.         job.setJarByClass(WeiboCount.class); 
    103.  
    104.         // Mapper 
    105.         job.setMapperClass(WeiBoMapper.class); 
    106.         // Mapper key输出类型 
    107.         job.setMapOutputKeyClass(Text.class); 
    108.         // Mapper value输出类型 
    109.         job.setMapOutputValueClass(Text.class); 
    110.          
    111.         // Reducer 
    112.         job.setReducerClass(WeiBoReducer.class); 
    113.         // Reducer key输出类型 
    114.         job.setOutputKeyClass(Text.class); 
    115.         // Reducer value输出类型 
    116.         job.setOutputValueClass(IntWritable.class); 
    117.          
    118.         // 输入路径 
    119.         FileInputFormat.addInputPath(job, new Path(args[0])); 
    120.         // 输出路径 
    121.         FileOutputFormat.setOutputPath(job, new Path(args[1])); 
    122.          
    123.         // 自定义输入格式 
    124.         job.setInputFormatClass(WeiboInputFormat.class) ; 
    125.         //自定义文件输出类别 
    126.         MultipleOutputs.addNamedOutput(job, FAN, TextOutputFormat.class, Text.class, IntWritable.class); 
    127.         MultipleOutputs.addNamedOutput(job, FOLLOWERS, TextOutputFormat.class, Text.class, IntWritable.class); 
    128.         MultipleOutputs.addNamedOutput(job, MICROBLOGS, TextOutputFormat.class, Text.class, IntWritable.class); 
    129.          
    130.         // 去掉job设置outputFormatClass，改为通过LazyOutputFormat设置   
    131.         LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);  
    132.          
    133.          //提交任务   
    134.         return job.waitForCompletion(true)?0:1; 
    135.     } 
    136.      
    137.     // 对Map内的数据进行排序（只适合小数据量） 
    138.     @SuppressWarnings("unchecked") 
    139.     public static Entry[] getSortedHashtableByValue(Map h) {   
    140.         Entry[] entries = (Entry[]) h.entrySet().toArray(new Entry[0]);   
    141.         // 排序 
    142.         Arrays.sort(entries, new Comparator>() { 
    143.             public int compare(Entry entry1, Entry entry2) { 
    144.                 return entry2.getValue().compareTo(entry1.getValue()); 
    145.             }  
    146.         }); 
    147.         return entries;   
    148.     } 
    149.      
    150.     public static void main(String[] args) throws Exception { 
    151.         String[] args0 = { 
    152.                 "hdfs://ljc:9000/buaa/microblog/weibo.txt", 
    153.                 "hdfs://ljc:9000/buaa/microblog/out/"  
    154.         }; 
    155.         int ec = ToolRunner.run(new Configuration(), new WeiboCount(), args0); 
    156.         System.exit(ec); 
    157.     } 
    158. } 
    

    5、运行结果
    

    

     
    

                 
            
            
            
    
                
                            
    
            
    
                
    
                    上一篇：IBM FlashSystem A9000R：显著提高大规模数据处理的性能
    
                
    
                
    
                    下一篇：物联网VSWiFi  两强争霸还是携手并进？                
    
            
    
            
            
    
                
                
    相关内容
    
                
            
    
            
    
            
            
    
                
            
    


        
        
    
    

    
    
    
        
        
    
            
            
    
                
                
    
                    
                    
    
                        
                        热门资讯
                        
                    
    
                    
                
    
                
            
    
            
            
            
    

                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        PHP新手之PHP入门
                        PHP是一种易于学习和使用的服务器端脚本语言。只需要很少的编程知识你就能使用PHP建立一个真正交互的...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        网络中立的未来 网络中立性是什...
                        《牛津词典》中对“网络中立”的解释是“电信运营商应秉持的一种原则，即不考虑来源地提供所有内容和应用的...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        各种千兆交换机的数据接口类型详...
                        千兆交换机有很多值得学习的地方，这里我们主要介绍各种千兆交换机的数据接口类型，作为局域网的主要连接设...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        粉嫩如何诠释霸道 东芝M805...
                        “霸道粉”是个什么玩意东芝M805拿过来的时候，笔者扑哧笑了，不是笑这款笔记本，而是笑这款产品的颜色...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        如何利用交换机和端口设置来管理...
                        在网络管理中，总是有些人让管理员头疼。下面我们就将介绍一下一个网管员利用交换机以及端口设置等来进行D...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        什么是大数据安全 什么是大数据...
                        在《为什么需要大数据安全分析》一文中，我们已经阐述了一个重要观点，即：安全要素信息呈现出大数据的特征...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        全面诠释网络负载均衡
                        负载均衡的出现大大缓解了服务器的压力，更是有效的利用了资源，提高了效率。那么我们现在来说一下网络负载...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        如何允许远程连接到MySQL数...
                        [[277004]]【51CTO.com快译】默认情况下，MySQL服务器仅侦听来自localhos...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        30分钟搞定iOS自定义相机
                        最近公司的项目中用到了相机，由于不用系统的相机，UI给的相机切图，必须自定义才可以。就花时间简单研究...
                    
    
                    
                
    
                                
                
    
                    
                    
    
                        
    
                    
    
                    
                    
                    
    
                        Windows恶意软件20年“...
                        在Windows的早期年代，病毒游走于系统之间，偶尔删除文件（但被删除的文件几乎都是可恢复的），并弹...