MaxCompute用户指南：MapReduce：示例程序：使用Counter示例
本示例中定义了三个 Counter：map_outputs，reduce_outputs 和 global_counts。您可以在Map/Reduce 的 setup，map/reduce 及 cleanup 接口中，获取任意自定义 Counter，并进行操作。
测试准备
准备好测试程序的 Jar 包，假设名字为 mapreduce-examples.jar，本地存放路径为 data\resources。
准备好 UserDefinedCounters 测试表和资源。
创建测试表。create table wc_in (key string, value string);- create table wc_out(key string, cnt bigint);
添加测试资源。
- add jar data\resources\mapreduce-examples.jar -f;
使用 tunnel 导入数据。
- tunnel upload data wc_in;
导入 wc_in 表的数据文件 data 的内容，如下所示：
- hello,odps
测试步骤
在 odpscmd 中执行 UserDefinedCounters，如下所示：
- jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
- com.aliyun.odps.mapred.open.example.UserDefinedCounters wc_in wc_out
预期结果
作业成功结束后，可以看到 Counters 的输出，如下所示：
- Counters: 3
- com.aliyun.odps.mapred.open.example.UserDefinedCounters$MyCounter
- MAP_TASKS=1
- REDUCE_TASKS=1
- TOTAL_TASKS=2
输出表 wc_out 中的内容，如下所示：
- +------------+------------+
- | key | cnt |
- +------------+------------+
- | hello | 1 |
- | odps | 1 |
- +------------+------------+
代码示例
- package com.aliyun.odps.mapred.open.example;
- import java.io.IOException;
- import java.util.Iterator;
- import com.aliyun.odps.counter.Counter;
- import com.aliyun.odps.counter.Counters;
- import com.aliyun.odps.data.Record;
- import com.aliyun.odps.mapred.JobClient;
- import com.aliyun.odps.mapred.MapperBase;
- import com.aliyun.odps.mapred.ReducerBase;
- import com.aliyun.odps.mapred.RunningJob;
- import com.aliyun.odps.mapred.conf.JobConf;
- import com.aliyun.odps.mapred.utils.SchemaUtils;
- import com.aliyun.odps.mapred.utils.InputUtils;
- import com.aliyun.odps.mapred.utils.OutputUtils;
- import com.aliyun.odps.data.TableInfo;
- /**
- *
- * User Defined Counters
- *
- **/
- public class UserDefinedCounters {
- enum MyCounter {
- TOTAL_TASKS, MAP_TASKS, REDUCE_TASKS
- }
- public static class TokenizerMapper extends MapperBase {
- private Record word;
- private Record one;
- @Override
- public void setup(TaskContext context) throws IOException {
- super.setup(context);
- Counter map_tasks = context.getCounter(MyCounter.MAP_TASKS);
- Counter total_tasks = context.getCounter(MyCounter.TOTAL_TASKS);
- map_tasks.increment(1);
- total_tasks.increment(1);
- word = context.createMapOutputKeyRecord();
- one = context.createMapOutputValueRecord();
- one.set(new Object[] { 1L });
- }
- @Override
- public void map(long recordNum, Record record, TaskContext context)
- throws IOException {
- for (int i = 0; i < record.getColumnCount(); i++) {
- word.set(new Object[] { record.get(i).toString() });
- context.write(word, one);
- }
- }
- }
- public static class SumReducer extends ReducerBase {
- private Record result = null;
- @Override
- public void setup(TaskContext context) throws IOException {
- result = context.createOutputRecord();
- Counter reduce_tasks = context.getCounter(MyCounter.REDUCE_TASKS);
- Counter total_tasks = context.getCounter(MyCounter.TOTAL_TASKS);
- reduce_tasks.increment(1);
- total_tasks.increment(1);
- }
- @Override
- public void reduce(Record key, Iterator<Record> values, TaskContext context)
- throws IOException {
- long count = 0;
- while (values.hasNext()) {
- Record val = values.next();
- count += (Long) val.get(0);
- }
- result.set(0, key.get(0));
- result.set(1, count);
- context.write(result);
- }
- }
- public static void main(String[] args) throws Exception {
- if (args.length != 2) {
- System.err
- .println("Usage: TestUserDefinedCounters <in_table> <out_table>");
- System.exit(2);
- }
- JobConf job = new JobConf();
- job.setMapperClass(TokenizerMapper.class);
- job.setReducerClass(SumReducer.class);
- job.setMapOutputKeySchema(SchemaUtils.fromString("word:string"));
- job.setMapOutputValueSchema(SchemaUtils.fromString("count:bigint"));
- InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
- OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
- RunningJob rJob = JobClient.runJob(job);
- Counters counters = rJob.getCounters();
- long m = counters.findCounter(MyCounter.MAP_TASKS).getValue();
- long r = counters.findCounter(MyCounter.REDUCE_TASKS).getValue();
- long total = counters.findCounter(MyCounter.TOTAL_TASKS).getValue();
- System.exit(0);
- }
- }
展开
收起
问答标签：
版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
相关问答
