从 reducer 访问映射器的计数器答案

【问题标题】：Accessing a mapper's counter from a reducer从 reducer 访问映射器的计数器
【发布时间】：2017-08-30 01:56:37
【问题描述】：

我需要从减速器中的映射器访问计数器。这可能吗？如果有怎么做？

例如：我的映射器是：

public class CounterMapper extends Mapper<Text,Text,Text,Text> {

    static enum TestCounters { TEST }

    @Override
    protected void map(Text key, Text value, Context context)
                    throws IOException, InterruptedException {
        context.getCounter(TestCounters.TEST).increment(1);
        context.write(key, value);
    }
}

我的减速机是

public class CounterReducer extends Reducer<Text,Text,Text,LongWritable> {

    @Override
    protected void reduce(Text key, Iterable<Text> values, Context context)
                        throws IOException, InterruptedException {
        Counter counter = context.getCounter(CounterMapper.TestCounters.TEST);
        long counterValue = counter.getValue();
        context.write(key, new LongWritable(counterValue));
    }
}

counterValue 始终为 0。是我做错了什么还是这不可能？

【问题讨论】：

标签： java hadoop

【解决方案1】：

在Reducer的configure(JobConf)中，可以使用JobConf对象来查找reducer自己的job id。这样，您的 reducer 可以创建自己的 JobClient——即与 jobtracker 的连接——并查询该作业（或任何相关作业）的计数器。

// in the Reducer class...
private long mapperCounter;

@Override
public void configure(JobConf conf) {
    JobClient client = new JobClient(conf);
    RunningJob parentJob = 
        client.getJob(JobID.forName( conf.get("mapred.job.id") ));
    mapperCounter = parentJob.getCounters().getCounter(MAP_COUNTER_NAME);
}

现在您可以在 reduce() 方法本身中使用 mapperCounter。

您实际上需要在这里尝试捕获。我正在使用旧 API，但适应新 API 应该不难。

请注意，映射器的计数器都应该在任何 reducer 启动之前完成，因此与 Justin Thomas 的评论相反，我相信您应该得到准确的值（只要 reducer 不递增相同的计数器！）

【讨论】：

reducer 中没有来自 mapper 的计数器似乎违反直觉，但在Hadoop reducer 可以在所有映射器完成之前开始执行。在那种情况下，计数器的值可能在减速器的不同时间读取不同。要了解更多关于如何在映射器完成执行之前启动减速器的信息，请访问这篇文章：stackoverflow.com/questions/11672676/…
@abhinavkulkarni 实际上，只有 reducer 的 shuffle 阶段可以在所有 mapper 启动之前启动，这与计数器无关。所以，当 reducer 的 reduce 阶段开始时，所有的 mapper 计数器都是正确的。来自同一篇文章：“另一方面，排序和归约只能在所有映射器完成后才能启动。”

【解决方案2】：

在新 API 上实施了 Jeff G 的解决方案：

    @Override
    public void setup(Context context) throws IOException, InterruptedException{
        Configuration conf = context.getConfiguration();
        Cluster cluster = new Cluster(conf);
        Job currentJob = cluster.getJob(context.getJobID());
        mapperCounter = currentJob.getCounters().findCounter(COUNTER_NAME).getValue();  
    }

【讨论】：

我试过了，但在以下行 mapperCounter = currentJob.getCounters().findCounter(COUNTER_NAME) 出现了 java 空点异常错误，我用自定义计数器替换了 COUNTER_NAME跨度>
似乎cluster.getJob(context.getJobID());在hadoop的独立操作中不起作用。在单节点集群模式下运行时，这对我有用。
你从哪里导入Cluster？ Intellij IDEA 建议我导入 org.apache.commons.math.stat.clustering.Cluster，仅此而已。而且这个导入不接受`Configuration作为构造函数的参数。

【解决方案3】：

map/reduce 的全部意义在于并行化作业。将有许多独特的映射器/缩减器，因此除了映射/缩减对的运行之外，该值无论如何都不正确。

他们有一个字数统计示例：

http://wiki.apache.org/hadoop/WordCount

您可以将 context.write(word,one) 更改为 context.write(line,one)

【讨论】：

【解决方案4】：

全局计数器值永远不会广播回每个映射器或减速器。如果您希望 reducer 可以使用映射器记录数，则需要依赖一些外部机制来执行此操作。

【讨论】：

Jobtracker 跟踪计数器。

【解决方案5】：

我问了this question，但我的问题还没有解决。但是，我想到了另一种解决方案。在映射器中，单词的数量被计算出来，并且可以在运行映射器的末尾的清理函数中以最小的键（以便该值在头部）写入中间输出。在 reducer 中，单词的数量是通过在 head 中添加值来计算的。下面提供了示例代码及其部分输出。

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.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * Created by tolga on 1/26/16.
 */
public class WordCount {
    static enum TestCounters { TEST }
    public static class Map extends Mapper<Object, Text, Text, LongWritable> {
        private final static LongWritable one = new LongWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(line);
            while (tokenizer.hasMoreTokens()) {
                word.set(tokenizer.nextToken());
                context.write(word, one);
                context.getCounter(TestCounters.TEST).increment(1);
            }
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            context.write(new Text("!"),new LongWritable(context.getCounter(TestCounters.TEST).getValue()));
        }
    }

    public static class Reduce extends Reducer<Text, LongWritable, Text, LongWritable> {

        public void reduce(Text key, Iterable<LongWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (LongWritable val : values) {
                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, "WordCount");
        job.setJarByClass(WordCount.class);

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

        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

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

        job.waitForCompletion(true);
    }
}

文本文件：

Turgut Özal University is a private university located in Ankara, Turkey. It was established in 2008 by the Turgut Özal Thought and Action Foundation and is named after former Turkish president Turgut Özal.

中间输出

**!	33**
2008	1
Action	1
Ankara,	1
Foundation	1
It	1
Thought	1
Turgut	1
Turgut	1
Turgut	1

**!	33**
2008	1
Action	1
Ankara,	1
Foundation	1
It	1
Thought	1
Turgut	3

【讨论】：

【解决方案6】：

从 itzhaki 的回答中得到改进

findCounter(COUNTER_NAME) 不再受支持 - https://hadoop.apache.org/docs/r2.7.0/api/org/apache/hadoop/mapred/Counters.html

@Override
public void setup(Context context) throws IOException, InterruptedException{
    Configuration conf = context.getConfiguration();
    Cluster cluster = new Cluster(conf);
    Job currentJob = cluster.getJob(context.getJobID());
    mapperCounter = currentJob.getCounters().findCounter(GROUP_NAME, COUNTER_NAME).getValue();  
}

GROUP_NAME 在调用计数器时指定。例如

context.getCounter("com.example.mycode", "MY_COUNTER").increment(1);

然后

mapperCounter = currentJob.getCounters().findCounter("com.example.mycode", "MY_COUNTER").getValue();

另外，重要的一点是，如果计数器不存在，它将用值 0 初始化一个。

【讨论】：