UserDefinedCounters example-MaxCompute(MaxCompute)-阿里云帮助中心

User-defined counters let you track custom metrics—such as the number of map tasks and reduce tasks—across a MapReduce job. MaxCompute aggregates counter values globally as the job runs, so you can read the final totals after the job completes. Counters are declared as Java enum values, incremented inside Mapper and Reducer methods, and retrieved through the RunningJob object.

This example demonstrates all three counter operations:

Operation	API	Where called
Define	Declare an `enum` with counter names	Class level
Increment	`context.getCounter()`, `Counter.increment()`	`setup()` methods
Retrieve	`RunningJob.getCounters()`, `Counters.findCounter()`	`main()` after job completes

Prerequisites

Before you begin, ensure that you have:

Completed the environment setup described in Getting started
The mapreduce-examples.jar package, located in the data\resources directory under the MaxCompute client bin directory

Prepare tables and resources

Create the input and output tables.

CREATE TABLE wc_in (key STRING, value STRING);
CREATE TABLE wc_out(key STRING, cnt BIGINT);

Add the JAR package as a resource.
```
add jar data\resources\mapreduce-examples.jar -f;
```
You can ignore the -f flag the first time you add the JAR package.
Import the test data using MaxCompute Tunnel. Run the following command from the MaxCompute client bin directory.
```
tunnel upload data.txt wc_in;
```
This imports the following record into wc_in:
```
hello,odps
```

Run the job

On the MaxCompute client, run the UserDefinedCounters class against the input and output tables.

jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar com.aliyun.odps.mapred.open.example.UserDefinedCounters wc_in wc_out

Expected output

After the job completes, three user-defined counters are reported:

Counters: 3
com.aliyun.odps.mapred.open.example.UserDefinedCounters$MyCounter
MAP_TASKS=1
REDUCE_TASKS=1
TOTAL_TASKS=2

The wc_out table contains the following results:

+------------+------------+
| key        | cnt        |
+------------+------------+
| hello      | 1          |
| odps       | 1          |
+------------+------------+

Sample code

For Project Object Model (POM) dependency configuration, see Precautions.

The example covers all three counter phases in a single class:

Define: MyCounter enum declares TOTAL_TASKS, MAP_TASKS, and REDUCE_TASKS
Increment: TokenizerMapper.setup() increments MAP_TASKS and TOTAL_TASKS; SumReducer.setup() increments REDUCE_TASKS and TOTAL_TASKS
Retrieve: main() calls rJob.getCounters() then counters.findCounter(MyCounter.MAP_TASKS).getValue() to read each counter after the job finishes

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 {
    // Phase 1 - Define: declare counter names as an enum
    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);
            // Phase 2 - Increment: get counters and increment them in the Mapper setup
            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();
            // Phase 2 - Increment: get counters and increment them in the Reducer setup
            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);
        // Phase 3 - Retrieve: read counter values after the job completes
        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);
    }
}