Hadoop学习之路（十五）MapReduce的多Job串联和全局计数器

MapReduce 多 Job 串联

需求

一个稍复杂点的处理逻辑往往需要多个 MapReduce 程序串联处理，多 job 的串联可以借助 MapReduce 框架的 JobControl 实现

实例

以下有两个 MapReduce 任务，分别是 Flow 的 SumMR 和 SortMR，其中有依赖关系：SumMR 的输出是 SortMR 的输入，所以 SortMR 的启动得在 SumMR 完成之后

Configuration conf1 = new Configuration();
        Configuration conf2 = new Configuration();

        Job job1 = Job.getInstance(conf1);
        Job job2 = Job.getInstance(conf2);

        job1.setJarByClass(MRScore3.class);
        job1.setMapperClass(MRMapper3_1.class);
        //job.setReducerClass(ScoreReducer3.class);

        job1.setMapOutputKeyClass(IntWritable.class);
        job1.setMapOutputValueClass(StudentBean.class);
        job1.setOutputKeyClass(IntWritable.class);
        job1.setOutputValueClass(StudentBean.class);

        job1.setPartitionerClass(CoursePartitioner2.class);

        job1.setNumReduceTasks(4);

        Path inputPath = new Path("D:\\MR\\hw\\work3\\input");
        Path outputPath = new Path("D:\\MR\\hw\\work3\\output_hw3_1");

        FileInputFormat.setInputPaths(job1, inputPath);
        FileOutputFormat.setOutputPath(job1, outputPath);

        job2.setMapperClass(MRMapper3_2.class);
        job2.setReducerClass(MRReducer3_2.class);

        job2.setMapOutputKeyClass(IntWritable.class);
        job2.setMapOutputValueClass(StudentBean.class);
        job2.setOutputKeyClass(StudentBean.class);
        job2.setOutputValueClass(NullWritable.class);

        Path inputPath2 = new Path("D:\\MR\\hw\\work3\\output_hw3_1");
        Path outputPath2 = new Path("D:\\MR\\hw\\work3\\output_hw3_end");

        FileInputFormat.setInputPaths(job2, inputPath2);
        FileOutputFormat.setOutputPath(job2, outputPath2);

        JobControl control = new JobControl("Score3");

        ControlledJob aJob = new ControlledJob(job1.getConfiguration());
        ControlledJob bJob = new ControlledJob(job2.getConfiguration());
        // 设置作业依赖关系
        bJob.addDependingJob(aJob);

        control.addJob(aJob);
        control.addJob(bJob);

        Thread thread = new Thread(control);
        thread.start();

        while(!control.allFinished()) {
            thread.sleep(1000);
        }
        System.exit(0);
        

MapReduce 全局计数器

MapReduce计数器是什么？

计数器是用来记录job的执行进度和状态的。它的作用可以理解为日志。我们可以在程序的某个位置插入计数器，记录数据或者进度的变化情况。

MapReduce计数器能做什么？

MapReduce 计数器（Counter）为我们提供一个窗口，用于观察 MapReduce Job 运行期的各种细节数据。对MapReduce性能调优很有帮助，MapReduce性能优化的评估大部分都是基于这些 Counter 的数值表现出来的。

MapReduce 都有哪些内置计数器？

MapReduce 自带了许多默认Counter，现在我们来分析这些默认 Counter 的含义，方便大家观察 Job 结果，如输入的字节数、输出的字节数、Map端输入/输出的字节数和条数、Reduce端的输入/输出的字节数和条数等。下面我们只需了解这些内置计数器，知道计数器组名称（groupName）和计数器名称（counterName），以后使用计数器会查找groupName和counterName即可。

1、任务计数器

在任务执行过程中，任务计数器采集任务的相关信息，每个作业的所有任务的结果会被聚集起来。例如，MAP_INPUT_RECORDS 计数器统计每个map任务输入记录的总数，并在一个作业的所有map任务上进行聚集，使得最终数字是整个作业的所有输入记录的总数。任务计数器由其关联任务维护，并定期发送给TaskTracker，再由TaskTracker发送给 JobTracker。因此，计数器能够被全局地聚集。下面我们分别了解各种任务计数器。

　　1）MapReduce 任务计数器

MapReduce 任务计数器的 groupName为org.apache.hadoop.mapreduce.TaskCounter，它包含的计数器如下表所示

　　2）文件系统计数器

文件系统计数器的 groupName为org.apache.hadoop.mapreduce.FileSystemCounter，它包含的计数器如下表所示

　　3）FileInputFormat 计数器

FileInputFormat 计数器的 groupName为org.apache.hadoop.mapreduce.lib.input.FileInputFormatCounter，它包含的计数器如下表所示，计数器名称列的括号（）内容即为counterName

　　4）FileOutputFormat 计数器

FileOutputFormat 计数器的 groupName为org.apache.hadoop.mapreduce.lib.input.FileOutputFormatCounter，它包含的计数器如下表所示

2、作业计数器

作业计数器由 JobTracker（或者 YARN）维护，因此无需在网络间传输数据，这一点与包括 “用户定义的计数器” 在内的其它计数器不同。这些计数器都是作业级别的统计量，其值不会随着任务运行而改变。 作业计数器计数器的 groupName为org.apache.hadoop.mapreduce.JobCounter，它包含的计数器如下表所示

计数器的该如何使用？

下面我们来介绍如何使用计数器。

1、定义计数器

1)枚举声明计数器

// 自定义枚举变量Enum
Counter counter = context.getCounter(Enum enum)

2)自定义计数器

/ 自己命名groupName和counterName
Counter counter = context.getCounter(String groupName,String counterName)

2、为计数器赋值

1)初始化计数器

counter.setValue(long value);// 设置初始值

　2)计数器自增

counter.increment(long incr);// 增加计数

3、获取计数器的值

　　1) 获取枚举计数器的值

Configuration conf = new Configuration();
Job job = new Job(conf, "MyCounter");
job.waitForCompletion(true);
Counters counters=job.getCounters();
Counter counter=counters.findCounter(LOG_PROCESSOR_COUNTER.BAD_RECORDS_LONG);// 查找枚举计数器，假如Enum的变量为BAD_RECORDS_LONG
long value=counter.getValue();//获取计数值

　　2) 获取自定义计数器的值

Configuration conf = new Configuration();
Job job = new Job(conf, "MyCounter");
job.waitForCompletion(true);
Counters counters = job.getCounters();
Counter counter=counters.findCounter("ErrorCounter","toolong");// 假如groupName为ErrorCounter，counterName为toolong
long value = counter.getValue();// 获取计数值

　　3) 获取内置计数器的值

Configuration conf = new Configuration();
Job job = new Job(conf, "MyCounter");
job.waitForCompletion(true);
Counters counters=job.getCounters();
// 查找作业运行启动的reduce个数的计数器，groupName和counterName可以从内置计数器表格查询（前面已经列举有）
Counter counter=counters.findCounter("org.apache.hadoop.mapreduce.JobCounter","TOTAL_LAUNCHED_REDUCES");// 假如groupName为org.apache.hadoop.mapreduce.JobCounter，counterName为TOTAL_LAUNCHED_REDUCES
long value=counter.getValue();// 获取计数值

　　4) 获取所有计数器的值

Configuration conf = new Configuration();
Job job = new Job(conf, "MyCounter");
Counters counters = job.getCounters();
for (CounterGroup group : counters) {
  for (Counter counter : group) {
    System.out.println(counter.getDisplayName() + ": " + counter.getName() + ": "+ counter.getValue());
  }
}

原文地址：https://www.cnblogs.com/qingyunzong/p/8576640.html