首页 > 代码库 > 大数据实战：用户流量分析系统

大数据实战：用户流量分析系统

2024-09-29 23:28:39   217人阅读

文章出处：http://blog.csdn.net/sdksdk0/article/details/51628874

作者：朱培

---------------------------------------------------------------------------------------------------------------

 

本文是结合Hadoop中的mapreduce来对用户数据进行分析，统计用户的手机号码、上行流量、下行流量、总流量的信息，同时可以按照总流量大小对用户进行分组排序等。是一个非常简洁易用的hadoop项目，主要用户进一步加强对MapReduce的理解及实际应用。文末提供源数据采集文件和系统源码。

本案例非常适合hadoop初级人员学习以及想入门大数据、云计算、数据分析等领域的朋友进行学习。

一、待分析的数据源

以下是一个待分析的文本文件，里面有非常多的用户浏览信息，保扩用户手机号码，上网时间，机器序列号，访问的IP，访问的网站，上行流量，下行流量，总流量等信息。这里只截取一小段，具体文件在文末提供下载链接。

技术分享


二、基本功能实现

想要统计出用户的上行流量、下行流量、总流量信息，我们需要建立一个bean类来对数据进行封装。于是新建应该Java工程，导包，或者直接建立一个MapReduce工程。在这里面建立一个FlowBean.java文件。
 
[html] view plain copy
 
 print?
  1.        private long upFlow;  
  2. private long dFlow;  
  3. private long sumFlow;  
然后就是各种右键生成get,set方法，还要toString(),以及生成构造函数，（千万记得要生成一个空的构造函数，不然后面进行分析的时候会报错）。
完整代码如下：
[java] view plain copy
 
 print?
  1. package cn.tf.flow;  
  2.   
  3. import java.io.DataInput;  
  4. import java.io.DataOutput;  
  5. import java.io.IOException;  
  6.   
  7. import org.apache.hadoop.io.Writable;  
  8. import org.apache.hadoop.io.WritableComparable;  
  9.   
  10. public class FlowBean  implements WritableComparable<FlowBean>{  
  11.       
  12.     private long upFlow;  
  13.     private long dFlow;  
  14.     private long sumFlow;  
  15.     public long getUpFlow() {  
  16.         return upFlow;  
  17.     }  
  18.     public void setUpFlow(long upFlow) {  
  19.         this.upFlow = upFlow;  
  20.     }  
  21.     public long getdFlow() {  
  22.         return dFlow;  
  23.     }  
  24.     public void setdFlow(long dFlow) {  
  25.         this.dFlow = dFlow;  
  26.     }  
  27.     public long getSumFlow() {  
  28.         return sumFlow;  
  29.     }  
  30.     public void setSumFlow(long sumFlow) {  
  31.         this.sumFlow = sumFlow;  
  32.     }  
  33.     public FlowBean(long upFlow, long dFlow) {  
  34.         super();  
  35.         this.upFlow = upFlow;  
  36.         this.dFlow = dFlow;  
  37.         this.sumFlow = upFlow+dFlow;  
  38.     }  
  39.     @Override  
  40.     public void readFields(DataInput in) throws IOException {  
  41.         upFlow=in.readLong();  
  42.         dFlow=in.readLong();  
  43.         sumFlow=in.readLong();  
  44.           
  45.     }  
  46.     @Override  
  47.     public void write(DataOutput out) throws IOException {  
  48.         out.writeLong(upFlow);  
  49.         out.writeLong(dFlow);  
  50.         out.writeLong(sumFlow);  
  51.     }  
  52.     public FlowBean() {  
  53.         super();  
  54.     }  
  55.   
  56.     @Override  
  57.     public String toString() {  
  58.            
  59.         return  upFlow + "\t" + dFlow + "\t" + sumFlow;  
  60.     }  
  61.     @Override  
  62.     public int compareTo(FlowBean o) {  
  63.           
  64.         return this.sumFlow>o.getSumFlow() ? -1:1;  
  65.     }  
  66.       
  67.       
  68.   
  69. }  

然后就是这个统计的代码了，新建一个FlowCount.java.在这个类里面，我直接把Mapper和Reduce写在同一个类里面了，如果按规范的要求应该是要分开写的。
在mapper中，获取后面三段数据的值，所以我的这里length-2,length-3.
[java] view plain copy
 
 print?
  1.       public static class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean> {  
  2.     @Override  
  3.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  
  4.   
  5.         // 拿到这行的内容转成string  
  6.         String line = value.toString();  
  7.   
  8.         String[] fields = StringUtils.split(line, "\t");  
  9.         try {  
  10.             if (fields.length > 3) {  
  11.                 // 获得手机号及上下行流量字段值  
  12.                 String phone = fields[1];  
  13.                 long upFlow = Long.parseLong(fields[fields.length - 3]);  
  14.                 long dFlow = Long.parseLong(fields[fields.length - 2]);  
  15.   
  16.                 // 输出这一行的处理结果,key为手机号，value为流量信息bean  
  17.                 context.write(new Text(phone), new FlowBean(upFlow, dFlow));  
  18.             } else {  
  19.                 return;  
  20.             }  
  21.         } catch (Exception e) {  
  22.   
  23.         }  
  24.   
  25.     }  
  26.   
  27. }  



 

在reduce中队数据进行整理，统计


 








[java] view plain copy

 

 print?




    

  1. public static class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean> {  
    2. 

  2.   
    3. 

  3.         @Override  
    4. 

  4.         protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {  
    5. 

  5.   
    6. 

  6.             long upSum = 0;  
    7. 

  7.             long dSum = 0;  
    8. 

  8.   
    9. 

  9.             for (FlowBean bean : values) {  
    10. 

  10.   
    11. 

  11.                 upSum += bean.getUpFlow();  
    12. 

  12.                 dSum += bean.getdFlow();  
    13. 

  13.             }  
    14. 

  14.   
    15. 

  15.             FlowBean resultBean = new FlowBean(upSum, dSum);  
    16. 

  16.             context.write(key, resultBean);  
    17. 

  17.   
    18. 

  18.         }  
    19. 

  19.   
    20. 

  20.     }  
    21. 






最后在main方法中调用执行。








[java] view plain copy

 


 print?





    

  1. public static void main(String[] args) throws Exception {  
    2. 

  2.   
    3. 

  3.         Configuration conf = new Configuration();  
    4. 

  4.         Job job = Job.getInstance(conf);  
    5. 

  5.   
    6. 

  6.         job.setJarByClass(FlowCount.class);  
    7. 

  7.   
    8. 

  8.         job.setMapperClass(FlowCountMapper.class);  
    9. 

  9.         job.setReducerClass(FlowCountReducer.class);  
    10. 

  10.   
    11. 

  11.         job.setMapOutputKeyClass(Text.class);  
    12. 

  12.         job.setMapOutputValueClass(FlowBean.class);  
    13. 

  13.   
    14. 

  14.         job.setOutputKeyClass(Text.class);  
    15. 

  15.         job.setOutputValueClass(FlowBean.class);  
    16. 

  16.   
    17. 

  17.         FileInputFormat.setInputPaths(job, new Path(args[0]));  
    18. 

  18.         FileOutputFormat.setOutputPath(job, new Path(args[1]));  
    19. 

  19.   
    20. 

  20.         boolean res = job.waitForCompletion(true);  
    21. 

  21.         System.exit(res ? 0 : 1);  
    22. 

  22.   
    23. 

  23.     }  
    24. 







当然啦，还需要先在你的hdfs根目录中建立/flow/data数据，然后我那个用户的数据源上传上去。








[java] view plain copy

 


 print?





    

  1. bin/hadoop fs -mkdir -p /flow/data  
    2. 

  2. bin/hadoop fs -put HTTP_20130313143750.dat /flow/data  
    3. 

  3. bin/hadoop jar  ../lx/flow.jar  
    4. 




 


把上面这个MapReduce工程打包成一个jar文件，然后用hadoop来执行这个jar文件。例如我放在~/hadoop/lx/flow.jar，然后再hadoop安装目录中执行








[java] view plain copy

 


 print?





    

  1. bin/hadoop jar ../lx/flowsort.jar cn/tf/flow/FlowCount  /flow/data  /flow/output  
    2. 







 


最后执行结果如下：



技术分享

在这整过过程中，我们是有yarnchild的进程在执行的，如下图所示：当整个过程执行完毕之后yarnchild也会自动退出。


技术分享


三、按总流量从大到小排序


如果你上面这个基本操作以及完成了的话，按总流量排序就非常简单了。我们新建一个FlowCountSort.Java.


全部代码如下：


 






[java] view plain copy

 


 print?





    

  1. package cn.tf.flow;  
    2. 

  2.   
    3. 

  3. import java.io.IOException;  
    4. 

  4.   
    5. 

  5. import org.apache.commons.lang.StringUtils;  
    6. 

  6. import org.apache.hadoop.conf.Configuration;  
    7. 

  7. import org.apache.hadoop.fs.Path;  
    8. 

  8. import org.apache.hadoop.io.LongWritable;  
    9. 

  9. import org.apache.hadoop.io.Text;  
    10. 

  10. import org.apache.hadoop.mapreduce.Job;  
    11. 

  11. import org.apache.hadoop.mapreduce.Mapper;  
    12. 

  12. import org.apache.hadoop.mapreduce.Reducer;  
    13. 

  13. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
    14. 

  14. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;  
    15. 

  15.   
    16. 

  16. public class FlowCountSort {  
    17. 

  17.   
    18. 

  18. public static class FlowCountSortMapper extends Mapper<LongWritable, Text, FlowBean, Text>{  
    19. 

  19.           
    20. 

  20.         @Override  
    21. 

  21.         protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {  
    22. 

  22.               
    23. 

  23.             String line=value.toString();  
    24. 

  24.             String[] fields=StringUtils.split(line,"\t");  
    25. 

  25.               
    26. 

  26.             String phone=fields[0];  
    27. 

  27.             long upSum=Long.parseLong(fields[1]);  
    28. 

  28.             long dSum=Long.parseLong(fields[2]);  
    29. 

  29.               
    30. 

  30.             FlowBean sumBean=new FlowBean(upSum,dSum);  
    31. 

  31.               
    32. 

  32.             context.write(sumBean, new Text(phone));  
    33. 

  33.           
    34. 

  34.         }     
    35. 

  35. }  
    36. 

  36.   
    37. 

  37.     public static class FlowCountSortReducer extends Reducer<FlowBean, Text, Text, FlowBean>{  
    38. 

  38.           
    39. 

  39.         //进来的“一组”数据就是一个手机的流量bean和手机号  
    40. 

  40.         @Override  
    41. 

  41.         protected void reduce(FlowBean key, Iterable<Text> values, Context context) throws IOException, InterruptedException {  
    42. 

  42.       
    43. 

  43.             context.write(values.iterator().next(), key);  
    44. 

  44.         }  
    45. 

  45.     }  
    46. 

  46.   
    47. 

  47.     public static void main(String[] args) throws Exception {  
    48. 

  48.   
    49. 

  49.         Configuration conf = new Configuration();  
    50. 

  50.         Job job = Job.getInstance(conf);  
    51. 

  51.   
    52. 

  52.         job.setJarByClass(FlowCountSort.class);  
    53. 

  53.   
    54. 

  54.         job.setMapperClass(FlowCountSortMapper.class);  
    55. 

  55.         job.setReducerClass(FlowCountSortReducer.class);  
    56. 

  56.   
    57. 

  57.         job.setMapOutputKeyClass(FlowBean.class);  
    58. 

  58.         job.setMapOutputValueClass(Text.class);  
    59. 

  59.   
    60. 

  60.         job.setOutputKeyClass(Text.class);  
    61. 

  61.         job.setOutputValueClass(FlowBean.class);  
    62. 

  62.   
    63. 

  63.         FileInputFormat.setInputPaths(job, new Path(args[0]));  
    64. 

  64.         FileOutputFormat.setOutputPath(job, new Path(args[1]));  
    65. 

  65.   
    66. 

  66.         boolean res = job.waitForCompletion(true);  
    67. 

  67.         System.exit(res ? 0 : 1);  
    68. 

  68.   
    69. 

  69.     }  
    70. 

  70.       
    71. 

  71. }  
    72. 






这个主要就是使用了FlowBean.java中的代码来实现的，主要是继承了WritableComparable<FlowBean>接口来实现，然后重写了compareTo()方法。


 


 






[html] view plain copy

 


 print?





    

  1. @Override  
    2. 

  2.     public int compareTo(FlowBean o) {  
    3. 

  3.           
    4. 

  4.         return this.sumFlow>o.getSumFlow() ? -1:1;  
    5. 

  5.     }  
    6. 

  6.       
    7. 





按照同样的方法对这个文件打成jar包，然后使用hadoop的相关语句进行执行就可以了。


 


 






[java] view plain copy

 


 print?





    

  1. bin/hadoop jar ../lx/flowsort.jar cn/tf/flow/FlowCountSort  /flow/output  /flow/sortoutput  
    2. 





结果图：


 



技术分享



四、按用户号码区域进行分类


流量汇总之后的结果需要按照省份输出到不同的结果文件中，需要解决两个问题：


 1、如何让mr的最终结果产生多个文件： 原理：MR中的结果文件数量由reduce
  task的数量绝对，是一一对应的 做法：在代码中指定reduce task的数量
 
 
  2、如何让手机号进入正确的文件 原理：让不同手机号数据发给正确的reduce task，就进入了正确的结果文件
  要自定义MR中的分区partition的机制（默认的机制是按照kv中k的hashcode%reducetask数）
  做法：自定义一个类来干预MR的分区策略——Partitioner的自定义实现类


主要代码与前面的排序是非常类似的，只要在main方法中添加如下两行代码就可以了。


 






[java] view plain copy

 


 print?





    

  1.         //指定自定义的partitioner  
    2. 

  2. job.setPartitionerClass(ProvincePartioner.class);  
    3. 

  3.   
    4. 

  4. job.setNumReduceTasks(5);  
    5. 






这里我们需要新建一个ProvincePartioner.java来处理号码分类的逻辑。


 


 






[java] view plain copy

 


 print?





    

  1. public class ProvincePartioner extends Partitioner<Text, FlowBean>{  
    2. 

  2.       
    3. 

  3.       
    4. 

  4. private static HashMap<String, Integer> provinceMap = new HashMap<String, Integer>();  
    5. 

  5.       
    6. 

  6.     static {  
    7. 

  7.           
    8. 

  8.         provinceMap.put("135", 0);  
    9. 

  9.         provinceMap.put("136", 1);  
    10. 

  10.         provinceMap.put("137", 2);  
    11. 

  11.         provinceMap.put("138", 3);        
    12. 

  12.     }  
    13. 

  13.       
    14. 

  14.     @Override  
    15. 

  15.     public int getPartition(Text key, FlowBean value, int numPartitions) {  
    16. 

  16.   
    17. 

  17.         String prefix = key.toString().substring(0, 3);  
    18. 

  18.         Integer partNum = provinceMap.get(prefix);  
    19. 

  19.         if(partNum == null) partNum=4;  
    20. 

  20.           
    21. 

  21.         return partNum;  
    22. 

  22.     }  
    23. 

  23.   
    24. 

  24. }  
    25. 






执行方法和前面也是一样的。从执行的流程中我们可以看到这里启动了5个reduce task,因为我这里数据量比较小，所以只启动了一个map task。


 


技术分享


 


到这里，整个用户流量分析系统就全部结束了。关于大数据的更多内容，欢迎关注。点击左上角头像下方“点击关注".感谢您的支持！


 


 


数据源下载地址：http://download.csdn.net/detail/sdksdk0/9545935


源码项目地址：https://github.com/sdksdk0/HDFS_MapReduce


 
大数据实战：用户流量分析系统






























        

            
                
                  
                
                

                    联系

                    我们