MaxCompute用户指南：MapReduce：Java SDK：原生SDK概述

本文将会为您介绍较为常用的 MapReduce 核心接口。如果您使用 Maven，可以从 Maven 库 中搜索“odps-sdk-mapred”获取不同版本的 Java SDK，相关配置信息如下：

1. <dependency>
2.     <groupId>com.aliyun.odps</groupId>
3.     <artifactId>odps-sdk-mapred</artifactId>
4.     <version>0.26.2-public</version>
5. </dependency>

注意：
MaxCompute 的 TaskContext 接口中提供了 progress 功能，但此功能是在防止 Worker长时间运行未结束，被框架误认为超时而被杀的情况出现。此接口更类似于向框架发送心跳信息，并不是用来汇报 Worker 进度。MaxComputeMapReduce 默认 Worker 超时时间为 10 分钟（系统默认配置，不受用户控制），如果超过 10 分钟，Worker仍然没有向框架发送心跳（调用 progress 接口），框架会强制停止该 Worker，MapReduce 任务失败退出。因此，建议您在Mapper/Reducer 函数中，定期调用 progress 接口，防止框架认为 Worker 超时，误杀任务。

注意：

• 
  通常情况下，GroupingColumns 包含在 KeySortColumns 中，KeySortColumns 和 PartitionColumns 要包含在 Key 中。
• 
  在 Map 端，Mapper 输出的 Record 会根据设置的 PartitionColumns 计算哈希值，决定分配到哪个 Reducer，会根据 KeySortColumns 对 Record 进行排序。
• 
  在 Reduce 端，输入 Records 在按照 KeySortColumns 排序好后，会根据GroupingColumns 指定的列对输入的 Records 进行分组，即会顺序遍历输入的 Records，把GroupingColumns 所指定列相同的 Records 作为一次 reduce 函数调用的输入。

2.     public Builder addMapper(Class<? extends Mapper> mapper)
4.     public Builder addMapper(Class<? extends Mapper> mapper,
6.            Column[] keySchema, Column[] valueSchema, String[] sortCols,
8.            SortOrder[] order, String[] partCols,
10.            Class<? extends Partitioner> theClass, String[] groupCols)
12.     public Builder addReducer(Class<? extends Reducer> reducer)
14.     public Builder addReducer(Class<? extends Reducer> reducer,
16.            Column[] keySchema, Column[] valueSchema, String[] sortCols,
18.            SortOrder[] order, String[] partCols,
20.            Class<? extends Partitioner> theClass, String[] groupCols)
22.     public Builder setOutputKeySchema(Column[] keySchema)
24.     public Builder setOutputValueSchema(Column[] valueSchema)
26.     public Builder setOutputKeySortColumns(String[] sortCols)
28.     public Builder setOutputKeySortOrder(SortOrder[] order)
30.     public Builder setPartitionColumns(String[] partCols)
32.     public Builder setPartitionerClass(Class<? extends Partitioner> theClass)
34.     public Builder setOutputGroupingColumns(String[] cols)

2.     Job job = new Job();
6.     Pipeline pipeline = Pipeline.builder()
8.      .addMapper(TokenizerMapper.class)
10.      .setOutputKeySchema(
12.          new Column[] { new Column("word", OdpsType.STRING) })
14.      .setOutputValueSchema(
16.          new Column[] { new Column("count", OdpsType.BIGINT) })
18.      .addReducer(SumReducer.class)
20.      .setOutputKeySchema(
22.          new Column[] { new Column("count", OdpsType.BIGINT) })
24.      .setOutputValueSchema(
26.          new Column[] { new Column("word", OdpsType.STRING),
28.          new Column("count", OdpsType.BIGINT) })
30.      .addReducer(IdentityReducer.class).createPipeline();
34.     job.setPipeline(pipeline);  
36.     job.addInput(...)
38.     job.addOutput(...)
40.     job.submit();

注意：

• 
  建议您在使用 MR² 功能前，首先了解 MapReduce 的基础用法。  
• 
  JobConf 仅能够配置 Map 后接单 Reduce 的 MapReduce 任务。