513      阿里云

原生SDK介绍__Java SDK介绍_MapReduce_大数据计算服务-阿里云

在本小节，我们仅会对较为常用的MapReduce核心接口做简短介绍。使用Maven的用户可以从Maven库中搜索”odps-sdk-mapred”获取不同版本的Java SDK，相关配置信息：

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

MapperBase

主要函数接口：

ReducerBase

主要函数接口：

TaskContext

主要函数接口：

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

JobConf

主要函数接口：

备注:

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

JobClient

主要函数接口：

RunningJob

主要函数接口：

InputUtils

主要函数接口：

OutputUtils

主要函数接口：

Pipeline

Pipeline是MR²的主体类。可以通过Pipeline.builder构建一个Pipeline。Pipeline的主要接口如下：

    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)

    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();