编写 MApreduce 代码以计算记录数答案

【问题标题】：Writing MApreduce code for counting number of records编写 MApreduce 代码以计算记录数
【发布时间】：2015-12-14 05:07:36
【问题描述】：

我想编写一个 mapreduce 代码来计算给定 CSV 文件中的记录数。我不知道在 map 中做什么以及在 reduce 中做什么我应该如何解决这个问题，谁能提出一些建议？

【问题讨论】：

标签： hadoop mapreduce

【解决方案1】：

您的地图应该为读取的每条记录发出 1
你的组合器应该发出它得到的所有“1”的总和（每张地图的小计）
你的 reducer 应该发出总记录数

【讨论】：

【解决方案2】：

您的映射器必须发出一个固定键（只需使用值为“count”的文本）和一个固定值 1（与您在 wordcount 示例中看到的相同）。

然后只需使用 LongSumReducer 作为减速器。

您的工作的输出将是一条带有键“count”的记录，值是您要查找的记录数。

您可以选择（显着！）通过使用相同的 LongSumReducer 作为组合器来提高性能。

【讨论】：

感谢您的回答，能否告诉我 LongSumReducer 到底是什么以及它与普通减速器有何不同？
LongSumReducer 是一个简单的reducer 实现，可以完全满足您的需求。所以你不必自己写。我在答案中将 URL 添加到文档中。
你能告诉我如何使用 LongSumReducer 吗？你能提供我的代码 sn-p 来帮助我吗？
当您尝试自己编写此代码时，它是微不足道的。查看 Hadoop 教程 (developer.yahoo.com/hadoop/tutorial/module4.html) 或一本好的 Hadoop 书籍并尝试自己编写。这不是一个“别人做我的功课”的网站。

【解决方案3】：

希望我有一个比接受的答案更好的解决方案。

我们不是为每条记录发出 1，而是在 map() 中增加一个计数器，并在 cleanup() 中的每个映射任务之后发出增加的计数器。

可以减少中间读写。而reducer 只需要聚合几个值的列表。

public class LineCntMapper extends
  Mapper<LongWritable, Text, Text, IntWritable> {

 Text keyEmit = new Text("Total Lines");
 IntWritable valEmit = new IntWritable();
 int partialSum = 0;

 public void map(LongWritable key, Text value, Context context) {
  partialSum++;
 }

 public void cleanup(Context context) {
  valEmit.set(partialSum);

   context.write(keyEmit, valEmit);

 }
}

你可以找到完整的工作代码here

【讨论】：

优雅高效的解决方案。

【解决方案4】：

使用 job.getcounters() 检索作业完成后您为每条记录增加的值。如果您使用 java 编写 mapreduce 作业，请使用 enum 作为计数机制。

【讨论】：

【解决方案5】：

我只使用身份映射器和身份缩减器。

这是 Mapper.class 和 Reducer.class。然后只需阅读map input records

你真的不需要做任何编码来得到这个。

【讨论】：

【解决方案6】：

import java.io.IOException;

import java.util.*;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.*;

import org.apache.hadoop.mapred.*;

public class LineCount 

{
    public static class Map extends MapReduceBase implements
            Mapper<LongWritable, Text, Text, IntWritable> 

{
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text("Total Lines");

    public void map(LongWritable key, Text value,
            OutputCollector<Text, IntWritable> output,Reporter reporter)
            throws IOException 
    {
        output.collect(word, one);
    }
}

public static class Reduce extends MapReduceBase implements
        Reducer<Text, IntWritable, Text, IntWritable> {
    public void reduce(Text key, Iterator<IntWritable> values,
            OutputCollector<Text, IntWritable> output, Reporter reporter)
            throws IOException {
        int sum = 0;
        while (values.hasNext()) {
            sum += values.next().get();
        }
        output.collect(key, new IntWritable(sum));
    }
}

public static void main(String[] args) throws Exception {
    JobConf conf = new JobConf(LineCount.class);

    conf.setJobName("LineCount");
    conf.setNumReduceTasks(5);
    conf.setOutputKeyClass(Text.class);
    conf.setOutputValueClass(IntWritable.class);

    conf.setMapperClass(Map.class);
    conf.setCombinerClass(Reduce.class);
    conf.setReducerClass(Reduce.class);

    conf.setInputFormat(TextInputFormat.class);
    conf.setOutputFormat(TextOutputFormat.class);

    FileInputFormat.setInputPaths(conf, new Path(args[0]));
    FileOutputFormat.setOutputPath(conf, new Path(args[1]));

    JobClient.runJob(conf);
}
}

【讨论】：