处理简单的hadoop map reduce代码任务

Question

我是hadoop世界的新手，真的在为一项简单的任务而苦苦挣扎，却找不到实现它的方法。

我们有一个场景，其中有不同的客户，他们呼叫不同的人(与不同的移动运营商)。每个电话呼叫详细信息都有呼叫开始时间和日期、呼叫结束时间和日期、呼叫的各个接线员姓名。

我们有如下格式的输入文件:客户电话号码|通话开始时间|通话结束时间|

已进行呼叫的各个移动运营商

例如，输入文件如下：

9898765467| 03:14 12/10/2013 | 03:40 12/10/2013 | airtel 

9898765467| 06:20 12/10/2013 | 07:05 12/10/2013 | vodaphone

9899875321| 08:14 13/10/2013 | 08:40 13/10/2013 | idea

9899875321| 04:15 13/10/2013 | 04:50 13/10/2013 | reliance

9899875321| 09:14 13/10/2013 | 09:30 13/10/2013 | idea

9898765467| 10:20 12/10/2013 | 10:55 12/10/2013 | vodaphone

现在我们想知道每个日期哪个手机号码呼叫哪个移动运营商，通话时间有多长？

如在给定的例子中，在((7:05-6:20)+(10:55-10:20))=45 + 35 = 80 mins的总通话时间的情况下，在12/10/2013上两次呼叫给vodaphone运营商的9898765467移动号码

因此，手机号码9898765467的输出应如下所示：

Mobile number | Date  | Operator name | Talk Time

9898765467 | 12/10/2013  | vodaphone | 80 mins

So final output file for all mobile numbers should be like:

9898765467 | 12/10/2013 | vodaphone | 80 mins

9898765467 | 12/10/2013 | airtel     | 26 mins

9899875321 | 13/10/2013 | idea       | 42 mins 

9899875321 | 13/10/2013 | reliance   | 35 mins 

有没有人能建议或提供map reduce代码来完成这项任务？

Answer 1

首先，您需要确定作业的Keys和Values (Map-Reduce)。

与本例一样，您需要为每个mobileNumber-date-operator组合生成持续时间。

因此，每一行的映射器输出将类似于(键-上面的组合，值-该行的持续时间)。

并且你的reducer需要做summation of durations for all这样的唯一键(组合)。

请通过示例来理解其中的逻辑。

由于我主要关注逻辑部分，因此您可能需要根据业务需求修改string/date formatting和line splits/tokens。

package stackoverflow.examples;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
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;

public class CallStatsJob {

    public static class CallStatsMapper extends
            Mapper<Object, Text, Text, LongWritable> {
        private LongWritable duration;
        private Text key = new Text();
        private String mobileNumber, startTime, endTime, operator;

        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            String[] words = value.toString().split(" \\| ");

            mobileNumber = words[0];
            startTime = words[1];
            endTime = words[2];
            operator = words[3];
// for debugging            
//          System.out.println(mobileNumber);
//          System.out.println(startTime);
//          System.out.println(endTime);
//          System.out.println(operator);

            SimpleDateFormat sdf = new SimpleDateFormat("hh:mm dd/M/yyyy");
//          String dateInString = "03:40 12/10/2013";
            Date stDate, enDate;
            try {
                stDate = sdf.parse(startTime);
                enDate = sdf.parse(endTime);
                Long diff = enDate.getTime() - stDate.getTime();
                Long diffMinutes = diff / (60 * 1000);

                this.key = new Text(mobileNumber+"-"+stDate.getDate()+"-"+operator);
                duration = new LongWritable(diffMinutes);

                context.write(this.key, duration);
            } catch (ParseException e) {
                e.printStackTrace();
            }

        }

    }

    public static class CallStatsReducer extends
            Reducer<Text, LongWritable, Text, LongWritable> {
        public void reduce(Text key, Iterable<LongWritable> values,
                Context context) throws IOException, InterruptedException {
            Long sum = 0L;
            for (LongWritable val : values) {
                sum = sum + val.get();
            }
            context.write(key, new LongWritable(sum));

        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = new Job(conf, "Caller Statistics");
        job.setJarByClass(CallStatsJob.class);
        job.setMapperClass(CallStatsMapper.class);
        job.setReducerClass(CallStatsReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);

        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true)?0:1);

    }

}

映射器输出：(如果您设置0缩减，您将能够看到此输出)

9898765467-12-airtel    26
9898765467-12-vodaphone 45
9899875321-13-idea      26
9899875321-13-reliance  35
9899875321-13-idea      16
9898765467-12-vodaphone 35

减速机输出：(上述作业的常规输出)

9898765467-12-airtel    26
9898765467-12-vodaphone 80
9899875321-13-idea      42
9899875321-13-reliance  35

我相信这个例子为您提供了解决方案，也为您提供了进一步的理解。

Answer 2

使用WordCount Program作为参考。

将映射键设置为数字|日期|运算符

将映射值MAke为持续时间。(您可以找到开始时间和结束时间的差异)

因此，映射器到此为止。

在reducer中，只需汇总每个键的持续时间列表。

从减速器中发出结果。

Answer 3

你可以使用hive来实现这一点，而不需要map reduce代码。在此文件上创建配置单元表。

创建外部表呼叫记录(手机字符串、开始时间字符串、结束时间字符串、操作员字符串)行格式分隔字段，以'|‘行结尾，以'\n’location '‘tblproperties (“skip.header.line.count”=“1”)结尾；

通过计算开始时间和结束时间之间的差异，在表上创建视图create view as select。This将帮助您计算差值。