java - 使用 Hadoop 计数器 - 多个作业

Question

我正在使用 Hadoop 进行 mapreduce 项目。我目前有 3 个顺序工作。

我想使用 Hadoop 计数器，但问题是我想在第一个作业中进行实际计数，但在第三个作业的 reducer 中访问计数器值。

我怎样才能做到这一点？我应该在哪里定义 enum？我需要通过它扔第二份工作吗？它也有助于查看一些代码示例来执行此操作，因为我还找不到任何东西。

注意:我使用的是 Hadoop 2.7.2

编辑:我已经尝试过解释的方法 here它没有成功。我的情况不同，因为我想从不同的工作访问计数器。 (不是从映射器到 reducer )。

我尝试做的事情:第一份工作:

public static void startFirstJob(String inputPath, String outputPath) throws IOException, ClassNotFoundException, InterruptedException {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "wordCount");
    job.setJarByClass(WordCount.class);
    job.setMapperClass(WordCountMapper.class);
    job.setCombinerClass(WordCountReducer.class);
    job.setReducerClass(WordCountReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(LongWritable.class);
    job.setInputFormatClass(SequenceFileInputFormat.class);
    job.setOutputFormatClass(SequenceFileOutputFormat.class);
    FileInputFormat.addInputPath(job, new Path(inputPath));
    FileOutputFormat.setOutputPath(job, new Path(outputPath));
    job.waitForCompletion(true);
}

在不同的类中定义计数器枚举:

public class CountersClass {
    public static enum N_COUNTERS {
        SOMECOUNT
    }
}

尝试读取计数器:

Cluster cluster = new Cluster(context.getConfiguration());
Job job = cluster.getJob(JobID.forName("wordCount"));
Counters counters = job.getCounters();
CountersClass.N_COUNTERS mycounter = CountersClass.N_COUNTERS.valueOf("SOMECOUNT");
Counter c1 = counters.findCounter(mycounter);
long N_Count = c1.getValue();

Answer 1

经典的解决方案是将作业的计数器值放入您需要访问它的后续作业的配置中:

因此请确保在计数作业映射器/缩减器中正确地递增它:

context.getCounter(CountersClass.N_COUNTERS.SOMECOUNT).increment(1);

然后计算作业完成后:

job.waitForCompletion(true);

Counter someCount = job.getCounters().findCounter(CountersClass.N_COUNTERS.SOMECOUNT);

//put counter value into conf object of the job where you need to access it
//you can choose any name for the conf key really (i just used counter enum name here)
job2.getConfiguration().setLong(CountersClass.N_COUNTERS.SOMECOUNT.name(), someCount.getValue());

下一部分是在另一个作业的映射器/缩减器中访问它。只需覆盖设置()例如:

private long someCount;

@Override
protected void setup(Context context) throws IOException,
    InterruptedException {
  super.setup(context);
  this.someCount  = context.getConfiguration().getLong(CountersClass.N_COUNTERS.SOMECOUNT.name(), 0));
}