无法访问 MapReduce 的 reducer 类中的计数器

Question

我正在通过以下方式从映射器中增加一个计数器

public static class TokenizerMapper
extends Mapper<Object, Text, Text, FloatWritable>{  
public static enum MyCounters { TOTAL };  
context.getCounter(MyCounters.TOTAL).increment(1);  

。

我正在尝试通过以下方式在减速器类中获取此计数器的值。

@Override
public void setup(Context context) throws IOException     ,InterruptedException{  
Configuration conf = context.getConfiguration();  
Cluster cluster = new Cluster(conf);  
Job currentJob = cluster.getJob(context.getJobID());  
Counters counters = currentJob.getCounters();  
Counter counter =          counters.findCounter(TokenizerMapper.MyCounters.TOTAL); 

但是当我运行代码时，
它总是给出一个

java.lang.NullPointerException at the last line
cluster.getJob(context.getJobID())

它总是返回 null。

我尝试了其他方法来访问在 reducer 的映射器中递增的计数器，但没有成功。

有人可以向我解释到底是什么问题以及如何从 reducer 访问计数器。我需要总计数的值来计算单词的百分比。

这是我的驱动程序代码。

Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "word count");
job.setJarByClass(WordCount.class);
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(IntSumReducer.class);
job.setReducerClass(IntSumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FloatWritable.class);
job.setNumReduceTasks(1);
FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
System.exit(job.waitForCompletion(true) ? 0 : 1);

Answer 1

我正在使用 Hadoop 2.7.0。

您无需实例化集群，即可访问 reducer 中的计数器。

在我的映射器中，我有以下代码：

// Define a enum in the Mapper class 
enum CustomCounter {Total};

// In the map() method, increment the counter for each record
context.getCounter(CustomCounter.Total).increment(1);

在我的减速器中，我访问计数器如下：

Counter counter = context.getCounter(CustomCounter.Total);

它非常适合我。

以下是我的 maven 依赖项：

<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>2.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-mapreduce-client-core</artifactId>
        <version>2.7.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-core</artifactId>
        <version>1.2.1</version>
    </dependency>
</dependencies>

Answer 2

您应该在代码中使用 JobClient.getJob(conf).getcounters() 而不是创建新的集群。