MaxCompute:Overview

Add the SDK dependency

Search for odps-sdk-mapred in the Maven repository to find the latest version. Add the following dependency to your project:

<dependency>
    <groupId>com.aliyun.odps</groupId>
    <artifactId>odps-sdk-mapred</artifactId>
    <version>0.40.10-public</version>
</dependency>

Data types

MaxCompute MapReduce supports the following data types. The table below shows the mapping to Java types.

Classes overview

MapperBase

The framework calls the mapper lifecycle methods in this order: setup once at the start, map once per input record, and cleanup once at the end.

public class WordCountMapper extends MapperBase {
    @Override
    public void setup(TaskContext context) throws IOException {
        // Initialize resources (for example, load lookup tables)
    }

    @Override
    public void map(long key, Record record, TaskContext context) throws IOException {
        // Process each input record and emit key-value pairs.
        // MaxCompute uses (long key, Record record), not (KEYIN key, VALUEIN value).
        Record mapKey = context.createMapOutputKeyRecord();
        Record mapValue = context.createMapOutputValueRecord();
        // Populate mapKey and mapValue, then emit:
        context.write(mapKey, mapValue);
    }

    @Override
    public void cleanup(TaskContext context) throws IOException {
        // Release resources or emit final accumulated results
    }
}

ReducerBase

The framework calls the reducer lifecycle methods in this order: setup once at the start, reduce once per unique key group, and cleanup once at the end.

public class WordCountReducer extends ReducerBase {
    @Override
    public void setup(TaskContext context) throws IOException {
        // Initialize resources
    }

    @Override
    public void reduce(Record key, Iterator<Record> values, TaskContext context) throws IOException {
        // Aggregate all values for this key
        long count = 0;
        while (values.hasNext()) {
            Record val = values.next();
            count += val.getBigint(0);
        }
        Record output = context.createOutputRecord();
        output.set(0, key.getString(0));
        output.set(1, count);
        context.write(output);
    }

    @Override
    public void cleanup(TaskContext context) throws IOException {
        // Release resources
    }
}

TaskContext

TaskContext is passed to each lifecycle method and provides access to output tables, resources, and framework utilities.

Worker timeout

The default worker timeout is 10 minutes and cannot be changed. If a worker does not call progress() within 10 minutes, the framework terminates the worker and the map or reduce task fails. Call progress() periodically in long-running tasks to keep workers alive. The method sends a heartbeat only — it does not report task progress.

JobConf

JobConf holds all configuration for a MapReduce job, including mapper and reducer classes, key-value schemas, and resource declarations.

How the map and reduce stages distribute and group records

At the map stage, the framework computes a hash of each output record based on the partition key columns to determine which reducer receives it. Records are sorted by the sort columns before being sent.

At the reduce stage, records are grouped by the grouping columns. All records sharing the same grouping key are passed together to a single reduce() call.

The grouping columns are selected from the sort columns. The sort columns and partition key columns must exist in keys.

JobClient

RunningJob

InputUtils

OutputUtils

Pipeline (extended MapReduce model)

Pipeline is the entry point for the extended MapReduce model, which lets you chain multiple mappers and reducers in a single job. Use Pipeline.builder() to construct the pipeline.

Builder methods

public Builder addMapper(Class<? extends Mapper> mapper)
public Builder addMapper(Class<? extends Mapper> mapper,
       Column[] keySchema, Column[] valueSchema, String[] sortCols,
       SortOrder[] order, String[] partCols,
       Class<? extends Partitioner> theClass, String[] groupCols)
public Builder addReducer(Class<? extends Reducer> reducer)
public Builder addReducer(Class<? extends Reducer> reducer,
       Column[] keySchema, Column[] valueSchema, String[] sortCols,
       SortOrder[] order, String[] partCols,
       Class<? extends Partitioner> theClass, String[] groupCols)
public Builder setOutputKeySchema(Column[] keySchema)
public Builder setOutputValueSchema(Column[] valueSchema)
public Builder setOutputKeySortColumns(String[] sortCols)
public Builder setOutputKeySortOrder(SortOrder[] order)
public Builder setPartitionColumns(String[] partCols)
public Builder setPartitionerClass(Class<? extends Partitioner> theClass)
public Builder setOutputGroupingColumns(String[] cols)

Example

The following example chains one mapper and two reducers (TokenizerMapper → SumReducer → IdentityReducer):

Job job = new Job();
Pipeline pipeline = Pipeline.builder()
    .addMapper(TokenizerMapper.class)
    .setOutputKeySchema(
        new Column[] { new Column("word", OdpsType.STRING) })
    .setOutputValueSchema(
        new Column[] { new Column("count", OdpsType.BIGINT) })
    .addReducer(SumReducer.class)
    .setOutputKeySchema(
        new Column[] { new Column("count", OdpsType.BIGINT) })
    .setOutputValueSchema(
        new Column[] { new Column("word", OdpsType.STRING),
        new Column("count", OdpsType.BIGINT) })
    .addReducer(IdentityReducer.class).createPipeline();

job.setPipeline(pipeline);
job.addInput(...)
job.addOutput(...)
job.submit();

To chain a single mapper with one reducer, use JobConf instead of Pipeline. Get familiar with the standard MapReduce API before using the extended model.