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