hadoop MapReduce运营商案例关于用户基站停留数据统计

注

如果需要文件和代码的话可评论区留言邮箱，我给你发源代码

本文来自博客园，作者：Arway，转载请注明原文链接：https://www.cnblogs.com/cenjw/p/hadoop-mapReduce-operator-case.html

实验要求

统计每个用户在不同时段中各个基站的停留时间。
1.功能描述

用户的手机，连接到不同的基站会产生一条记录。
数据格式为：用户标识 设备标识 基站位置 通讯的日期 通讯时间
example： 0000009999 0054785806 00000089 2016-02-21 21:55:37
需要得到的数据格式为：
用户标识 时段 基站位置 停留时间
example： 0000000001 09-18 00000003 15
用户0000000001在09-18点这个时间段在基站00000003停留了15分钟
2.实现思路

程序运行支持传入时间段，比如“09-18-24”，表示分为0点到9点，9点到18点，18点到24点三个时间段。

• （1）Mapper阶段
  对输入的数据，算出它属于哪个时间段。
  k1：每行记录在文本中的偏移量。
  v2：一条记录
  k2用“用户ID，时间段”输出。
  v2用“基站位置,时间”。时间用unix time
  
• （2）Reducer阶段
  对获取的v3（v3是一个集合，每个元素是v2，相当于按照k2对v2分组）进行排序，以时间升序排序。
  计算两两之间的时间间隔，保存到另一个集合中，两个不同的时间间隔中，从基站A移动到基站B，这样获取到在A基站的停留的时间。
  同理从基站B移动到基站C，基站C移动到基站D，依次类推，所有的时间都获取到。再把时间累加起来，就可以获取到总的时间。
  

本文来自博客园，作者：Arway，转载请注明原文链接：https://www.cnblogs.com/cenjw/p/hadoop-mapReduce-operator-case.html

代码实现

PhoneMain.java

1. package phoneMapReduce;
3. import org.apache.hadoop.conf.Configuration;
4. import org.apache.hadoop.fs.Path;
5. import org.apache.hadoop.io.Text;
6. import org.apache.hadoop.mapreduce.Job;
7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
9. import java.io.IOException;
11. /**
12. * Created by ue50 on 11/13/19.
13. */
14. public class PhoneMain
15. {
16.     public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException
17.     {
18.         //String.equals()比较字符串的值是否相同
19.         if(args == null || "0".equals(args[0]))
20.         {
21.             throw new RuntimeException("argument is not right!");
22.         }
23.         //Configuration是作业的配置信息类
24.         Configuration configuration = new Configuration();
25.         //set(String name, String value)设置配置项
26.         configuration.set("timeRange", args[0]);
28.         Job job = Job.getInstance(configuration);
29.         job.setJarByClass(PhoneMain.class);
31.         job.setMapperClass(PhoneMapper.class);
32.         job.setMapOutputKeyClass(Text.class);
33.         job.setMapOutputKeyClass(Text.class);
35.         job.setReducerClass(PhoneReducer.class);
36.         job.setOutputKeyClass(Text.class);
37.         job.setOutputValueClass(Text.class);
39.         //FileInputFormat.setInputPaths(job, new Path("hdfs://xdata-m0:8020/user/ue50/pos.txt"));
40.         //FileOutputFormat.setOutputPath(job, new Path("hdfs://xdata-m0:8020/user/ue50/out"));
42.         FileInputFormat.setInputPaths(job, new Path(args[1]));
43.         FileOutputFormat.setOutputPath(job, new Path(args[2]));
45.         job.waitForCompletion(true);
46.     }
47. }

复制代码

Mapper阶段
PhoneMapper.java

1. package phoneMapReduce;
3. import org.apache.hadoop.conf.Configuration;
4. import org.apache.hadoop.io.LongWritable;
5. import org.apache.hadoop.io.Text;
6. import org.apache.hadoop.mapreduce.Mapper;
7. import java.io.IOException;
9. /**
10. * Created by ue50 on 11/13/19.
11. */
12. public class PhoneMapper extends Mapper<LongWritable, Text, Text, Text>
13. {
14.     private int[] timeRangeList;
15.     @Override
16.     //setup()被MapReduce框架仅且执行一次，在执行Map任务前，进行相关变量或者资源的集中初始化工作
17.     protected void setup(Context context) throws IOException,InterruptedException
18.     {
19.         //Configuration是作业的配置信息类,通过Configuration可以实现在多个mapper和多个reducer任务之间共享信息
20.         Configuration configuration = context.getConfiguration();
22.         //get(String name)根据配置项的键name获取相应的值
23.         String timeRange = configuration.get("timeRange");//运行时传入的时间段,比如“09-18-24”
24.         String[] timeRangeString = timeRange.split("-");
26.         timeRangeList = new int[timeRangeString.length];
27.         for(int i = 0; i < timeRangeString.length;i++)
28.         {
29.             //timeRangeList数组保存传入的时间,如:09、18、24
30.             timeRangeList[i] = Integer.parseInt(timeRangeString[i]);
31.         }
32.     }
34.     @Override
35.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
36.     {
37.         String values[] = value.toString().split("\\s+");//对一条记录"用户标识 设备标识   基站位置 通讯的时间"按空格拆分
38.         String userId = values[0];//用户标识
39.         String baseStation = values[2];//基站位置
40.         String timeString = values[4];//访问时间,如:21:55:37
42.         String[] times = timeString.split(":");//对访问时间按':'拆分
43.         int hour = Integer.parseInt(times[0]);//小时
45.         //startHour、endHour时间段的起止时间
46.         int startHour = 0;
47.         int endHour = 0;
48.         for(int i = 0; i < timeRangeList.length; i++)
49.         {
50.             if(hour < timeRangeList[i])
51.             {
52.                 if(i == 0)
53.                 {
54.                     startHour = 0;
55.                 }
56.                 else
57.                 {
58.                     startHour = timeRangeList[i-1];
59.                 }
60.                 endHour = timeRangeList[i];
61.                 break;
62.             }
63.         }
65.         if(startHour == 0 && endHour == 0)
66.         {
67.             return;
68.         }
70.         //k2:用户标识  时间段  v2:基站位置-访问时间
71.         context.write(new Text(userId + "\t" + startHour + "-" + endHour + "\t"), new Text(baseStation + "-" + timeString));
72.     }
73. }

复制代码

Reducer阶段
[code]package phoneMapReduce;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;import java.text.DateFormat;import java.text.ParseException;import java.text.SimpleDateFormat;import java.util.*;/** * Created by ue50 on 11/13/19. */public class PhoneReducer extends Reducer{    @Override    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException    {        List valueList = new LinkedList();//基于链表的动态数组        //Map是一种把键对象和值对象映射的集合,TreeMap是一个有序的key-value集合，        //它是通过红黑树实现的,TreeMap中的元素默认按照key的自然排序排列        Map residenceTimeMap = new TreeMap();        for(Text value : values)        {            String item = value.toString();            valueList.add(item);//"基站位置-访问时间"的集合        }        if(valueList == null || valueList.size()