hadoop-mr.txt

通过MR实现好友推荐
更新时间：2017-11-08 15:32:55


本页目录
实验介绍
操作步骤
社交网络是现如今影响力巨大的信息平台，社交网站中，您可以通过 可能感兴趣的人 途径增加交友方式。可能感兴趣的人 也称作 好友推荐， 它主要是通过查找两个非好友之间的共同好友情况来实现的。本文将通过一个示例，简单介绍如何通过 MapReduce 的方式实现好友推荐功能。

实验介绍
A，B，C，D，E 五个人的好友关系如下图所示，其中实线表示互为好友关系。那么，如何获取两个不是好友的两个人之间的好友数，并以此为参考，向用户推荐陌生人呢？

图片名称

主要通过以下几个步骤实现：

将好友关系分配到两个 Map 进行处理，其中每个 Map 包含 3 条好友关系。对每一条好友关系进行拆分，若 Key 中的两个人为朋友，则记录 value 值为0，否则 value 值为 1。将拆分的结果进行排序，其中（A B）和（B A）作为同一个 key（A B）。

图片名称

分别对两个 Map 处理的记录进行初步合并，若两个记录的 Key 值相同且每条记录的 Value 都不为 0，则 Value 值加 1。

注意：

在 Combine 阶段，必须保留 Value 为 0 的记录，否则，在 Reduce 阶段，获取的结果会出错。

图片名称

通过 Reduce 方式，合并两个 Map 处理的 Combine 结果。

若两个记录的 Key 值相同且每条记录的 Value 都不为 0，则 Value 值加 1。

将 Value 值为 0 的记录删除。

获取不为好友的两个用户之间的公共好友数：Key 为两个不为好友的用户，Value 是两个不是好友的用户之间的共同好友数。社交网站或者 APP 可以根据这个数值对不是好友的两个用户进行推荐。

图片名称

操作步骤
新建数据表
登录 DataWorks 管理控制台，单击相应项目空间后的 进入工作区。

单击顶部导航栏中的 数据开发，进入数据开发首页后单击 新建 > 新建脚本文件 或 新建脚本。

图片名称

配置新建脚本文件弹出框中的相关信息，填写文件名称，选择类型为 ODPS SQL 后，单击提交。如下图所示：

图片名称

输入建表语句，如下所示：

drop table if exists dual;--创建系统dual
create table dual(id bigint);--如project中不存在此伪表，则需创建并初始化数据
insert overwrite table dual select count(*)from dual;--向系统伪表初始化数据
---创建好友推荐MR的数据输入表.其中uid表示某个用户;friends表示uid用户的好友
create table friends_in (uid string, friends string);
---创建好友推荐MR的数据输出表.其中userA表示某个用户;userB表示不是userA的用户,cnt表示userA和userB之间的共同好友数。
create table friends_out (userA string, userB string, cnt bigint);
单击 运行，直至日志信息返回成功表示目标表创建成功。

图片名称

单击 保存，保存输入的 SQL 建表语句。

导入本地数据
单击顶部功能栏中的 导入 > 导入本地数据，打开本地保存的文件 friends_in_data.csv（点此下载）。

所有配置均设为默认，并查看导入的数据。完成后，单击 下一步。

注意：

在真实的工作环境中，数据必须以txt或csv的文件类型导入。

图片名称

在本地数据导入页面的 导入至表 中，输入 friends_in，即将本次实验的测试数据，导入到好友推荐的输入表 friends_in 中，确定 目标字段 与 源字段 匹配。完成后单击 导入。

图片名称

由于数据量较大，请等待1-2分钟。

数据导入完成后，可输入语句进行查询、确认。如下图所示：

1

添加 MR 资源
单击左侧导航栏中的 资源管理，单击列表右上角的 上传资源。

1

配置资源上传弹出框中的信息，选择需要上传的文件 Friends_MR。如下图所示：

图片名称

单击 提交。

在左侧导航栏的 资源管理 下，即可看到上传成功的 Jar 包 friends_mr.jar。

测试并验证好友推荐
单击顶部导航栏中的 新建 > 新建任务，开始创建本次实验的 MR 任务。

在弹出的对话框中，选择新建任务的 任务类型 为 节点任务，配置如下图所示：

图片名称

单击 创建。

在任务页面中输入各配置信息，如下图所示：

图片名称

配置项说明：

MRJar 包：单击文本框，选择 friends_mr.jar。
资源：默认设置为 friends_mr.jar。

输入表：输入 friends_in。

mapper：输入 friends_mr_odps.FriendsMapper，此为 Jar 包中 Mapper 的 class 全名。

reducer：输入 friends_mr_odps.FriendsReducer，此为 Jar 包中 Reducer 的 class 全名。

combiner：输入 friends_mr_odps.FriendsCombiner，此为 Jar 包中 Combiner 的 class 全名。

输出表：输入 friends_out。

输出 Key：输入 userA:String，userB:String。

输出 Val：输入 cnt:Bigint。

保存 并 运行 配置的 OPEN MR 任务，可在底部的 日志 中，查看运行状态和运行结果。如下图所示：

图片名称

在脚本文件中输入如下的 SQL 命令，并单击 运行，查询共同好友超过 2 个的数据信息。

SELECT * FROM friends_out WHERE cnt>2 order by cnt desc limit 100;








===================================
WordCount示例
更新时间：2019-03-07 16:31:37

编辑 ·
· 我的收藏
本页目录
测试准备
测试步骤
预期结果
代码示例
本文向您介绍MapReduce WordCount示例程序。

测试准备
准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
创建测试表。
试用
create table wc_in (key string, value string);
create table wc_out(key string, cnt bigint);
添加测试资源。
试用
add jar data\resources\mapreduce-examples.jar -f;
准备好WordCount测试表和资源。
使用tunnel导入数据。
试用
tunnel upload data wc_in;
导入wc_in表的数据文件data的内容，如下所示：
试用
hello,odps
测试步骤
在odpscmd中执行WordCount，如下所示：
试用
jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.WordCount wc_in wc_out
预期结果
作业成功结束后，输出表wc_out中的内容，如下所示：
试用
+------------+------------+
| key        | cnt        |
+------------+------------+
| hello      | 1          |
| odps       | 1          |
+------------+------------+
代码示例
试用
package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import java.util.Iterator;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.TableInfo;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.ReducerBase;
import com.aliyun.odps.mapred.TaskContext;
import com.aliyun.odps.mapred.conf.JobConf;
import com.aliyun.odps.mapred.utils.InputUtils;
import com.aliyun.odps.mapred.utils.OutputUtils;
import com.aliyun.odps.mapred.utils.SchemaUtils;
public class WordCount {
public static class TokenizerMapper extends MapperBase {
private Record word;
private Record one;
@Override
public void setup(TaskContext context) throws IOException {
word = context.createMapOutputKeyRecord();
one = context.createMapOutputValueRecord();
one.set(new Object[] { 1L });
System.out.println("TaskID:" + context.getTaskID().toString());
}
@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);
}
}
}
/**
* A combiner class that combines map output by sum them.
**/
public static class SumCombiner extends ReducerBase {
private Record count;
@Override
public void setup(TaskContext context) throws IOException {
count = context.createMapOutputValueRecord();
}
//combiner实现的接口和reducer一样，可以立即为在mapper本地执行的一个reduce，作用是减少mapper的输出量
@Override
public void reduce(Record key, Iterator<Record> values, TaskContext context)
throws IOException {
long c = 0;
while (values.hasNext()) {
Record val = values.next();
c += (Long) val.get(0);
}
count.set(0, c);
context.write(key, count);
}
}
/**
* A reducer class that just emits the sum of the input values.
**/
public static class SumReducer extends ReducerBase {
private Record result = null;
@Override
public void setup(TaskContext context) throws IOException {
result = context.createOutputRecord();
}
@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: WordCount <in_table> <out_table>");
System.exit(2);
}
JobConf job = new JobConf();
job.setMapperClass(TokenizerMapper.class);
job.setCombinerClass(SumCombiner.class);
job.setReducerClass(SumReducer.class);
//设置mapper中间结果的key和value的schema, mapper的中间结果输出也是record的形式
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);
JobClient.runJob(job);
}
}

MapOnly示例
更新时间：2018-12-12 10:52:47

编辑 ·
· 我的收藏
本页目录
测试准备
测试步骤
预期结果
代码示例
对于MapOnly的作业，Map直接将<Key,Value>信息输出到MaxCompute的表中，您只需要指定输出表即可，不需指定Map输出的Key/Value元信息。

测试准备
准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
准备好MapOnly的测试表和资源。
创建测试表。
试用
create table wc_in (key string, value string);
create table wc_out(key string, cnt bigint);
添加测试资源。
试用
add jar data\resources\mapreduce-examples.jar -f;
使用tunnel导入数据。
试用
tunnel upload data wc_in;
导入wc_in表的数据文件data的内容，如下所示：
试用
hello,odps
hello,odps
测试步骤
在odpscmd中执行MapOnly，如下所示：
试用
jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.MapOnly wc_in wc_out map
预期结果
作业成功结束后，输出表wc_out中的内容，如下所示：
试用
+------------+------------+
| key        | cnt        |
+------------+------------+
| hello      | 1          |
| hello      | 1          |
+------------+------------+
代码示例
试用
package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
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;
public class MapOnly {
public static class MapperClass extends MapperBase {
@Override
public void setup(TaskContext context) throws IOException {
  boolean is = context.getJobConf().getBoolean("option.mapper.setup", false);
  //Main函数在jobconf里设置了option.mapper.setup为true，才会执行下面的逻辑
  if (is) {
    Record result = context.createOutputRecord();
    result.set(0, "setup");
    result.set(1, 1L);
    context.write(result);
  }
}
@Override
public void map(long key, Record record, TaskContext context) throws IOException {
  boolean is = context.getJobConf().getBoolean("option.mapper.map", false);
  //Main函数在jobconf里设置了option.mapper.map为true，才会执行下面的逻辑
  if (is) {
    Record result = context.createOutputRecord();
    result.set(0, record.get(0));
    result.set(1, 1L);
    context.write(result);
  }
}
@Override
public void cleanup(TaskContext context) throws IOException {
  boolean is = context.getJobConf().getBoolean("option.mapper.cleanup", false);
  //Main函数在jobconf里设置了option.mapper.cleanup为true，才会执行下面的逻辑
  if (is) {
    Record result = context.createOutputRecord();
    result.set(0, "cleanup");
    result.set(1, 1L);
    context.write(result);
  }
}
}
public static void main(String[] args) throws Exception {
if (args.length != 2 && args.length != 3) {
  System.err.println("Usage: OnlyMapper <in_table> <out_table> [setup|map|cleanup]");
  System.exit(2);
}
JobConf job = new JobConf();
job.setMapperClass(MapperClass.class);
//对于MapOnly的作业，必须显式设置reducer的个数为0
job.setNumReduceTasks(0);
//设置输入输出的表信息
InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
if (args.length == 3) {
  String options = new String(args[2]);
//jobconf中可以设置自定义的key,value值，在mapper中通过context的getJobConf可以获取到相关的设置
  if (options.contains("setup")) {
    job.setBoolean("option.mapper.setup", true);
  }
  if (options.contains("map")) {
    job.setBoolean("option.mapper.map", true);
  }
  if (options.contains("cleanup")) {
    job.setBoolean("option.mapper.cleanup", true);
  }
}
JobClient.runJob(job);
}
}

-----------------------
多路输入输出示例

本页目录
测试准备
测试步骤
预期结果
代码示例
测试准备
准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
准备好多路输入输出的测试表和资源。
创建测试表。
试用
create table wc_in1(key string, value string);
create table wc_in2(key string, value string);
create table mr_multiinout_out1 (key string, cnt bigint);
create table mr_multiinout_out2 (key string, cnt bigint) partitioned by (a string, b string);
alter table mr_multiinout_out2 add partition (a='1', b='1');
alter table mr_multiinout_out2 add partition (a='2', b='2');
添加测试资源。
试用
add jar data\resources\mapreduce-examples.jar -f;
使用tunnel导入数据。
试用
tunnel upload data1 wc_in1;
tunnel upload data2 wc_in2;
导入wc_in1表的数据文件data的内容，如下所示：
试用
hello,odps
导入wc_in2表的数据文件data的内容，如下所示：
试用
hello,world
测试步骤
在odpscmd中执行MultipleInOut，如下所示：
试用
jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.MultipleInOut wc_in1,wc_in2 mr_multiinout_out1,mr_multiinout_out2|a=1/b=1|out1,mr_multiinout_out2|a=2/b=2|out2;
预期结果
作业成功结束后，mr_multiinout_out1中的内容，如下所示：
试用
+------------+------------+
| key        | cnt        |
+------------+------------+
| default    | 1          |
+------------+------------+
mr_multiinout_out2中内容，如下所示：
试用
+--------+------------+---+---+
| key    | cnt        | a | b |
+--------+------------+---+---+
| odps   | 1          | 1 | 1 |
| world  | 1          | 1 | 1 |
| out1   | 1          | 1 | 1 |
| hello  | 2          | 2 | 2 |
| out2   | 1          | 2 | 2 |
+--------+------------+---+---+
代码示例
试用
package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.data.TableInfo;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.ReducerBase;
import com.aliyun.odps.mapred.TaskContext;
import com.aliyun.odps.mapred.conf.JobConf;
import com.aliyun.odps.mapred.utils.InputUtils;
import com.aliyun.odps.mapred.utils.OutputUtils;
import com.aliyun.odps.mapred.utils.SchemaUtils;
/**
 * Multi input & output example.
 **/
public class MultipleInOut {
  public static class TokenizerMapper extends MapperBase {
    Record word;
    Record one;
    @Override
    public void setup(TaskContext context) throws IOException {
      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;
    private Record result1;
    private Record result2;
    @Override
    public void setup(TaskContext context) throws IOException {
      //对于不同的输出需要创建不同的record，通过label来区分
      result = context.createOutputRecord();
      result1 = context.createOutputRecord("out1");
      result2 = context.createOutputRecord("out2");
    }
    @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);
      }
      long mod = count % 3;
      if (mod == 0) {
        result.set(0, key.get(0));
        result.set(1, count);
        //不指定label，输出的默认(default)输出
        context.write(result);
      } else if (mod == 1) {
        result1.set(0, key.get(0));
        result1.set(1, count);
        context.write(result1, "out1");
      } else {
        result2.set(0, key.get(0));
        result2.set(1, count);
        context.write(result2, "out2");
      }
    }
    @Override
    public void cleanup(TaskContext context) throws IOException {
      Record result = context.createOutputRecord();
      result.set(0, "default");
      result.set(1, 1L);
      context.write(result);
      Record result1 = context.createOutputRecord("out1");
      result1.set(0, "out1");
      result1.set(1, 1L);
      context.write(result1, "out1");
      Record result2 = context.createOutputRecord("out2");
      result2.set(0, "out2");
      result2.set(1, 1L);
      context.write(result2, "out2");
    }
  }
  //将分区字符串如"ds=1/pt=2"转为map的形式
  public static LinkedHashMap<String, String> convertPartSpecToMap(
      String partSpec) {
    LinkedHashMap<String, String> map = new LinkedHashMap<String, String>();
    if (partSpec != null && !partSpec.trim().isEmpty()) {
      String[] parts = partSpec.split("/");
      for (String part : parts) {
        String[] ss = part.split("=");
        if (ss.length != 2) {
          throw new RuntimeException("ODPS-0730001: error part spec format: "
              + partSpec);
        }
        map.put(ss[0], ss[1]);
      }
    }
    return map;
  }
  public static void main(String[] args) throws Exception {
    String[] inputs = null;
    String[] outputs = null;
    if (args.length == 2) {
      inputs = args[0].split(",");
      outputs = args[1].split(",");
    } else {
      System.err.println("MultipleInOut in... out...");
      System.exit(1);
    }
    JobConf job = new JobConf();
    job.setMapperClass(TokenizerMapper.class);
    job.setReducerClass(SumReducer.class);
    job.setMapOutputKeySchema(SchemaUtils.fromString("word:string"));
    job.setMapOutputValueSchema(SchemaUtils.fromString("count:bigint"));
    //解析用户的输入表字符串
    for (String in : inputs) {
      String[] ss = in.split("\\|");
      if (ss.length == 1) {
        InputUtils.addTable(TableInfo.builder().tableName(ss[0]).build(), job);
      } else if (ss.length == 2) {
        LinkedHashMap<String, String> map = convertPartSpecToMap(ss[1]);
        InputUtils.addTable(TableInfo.builder().tableName(ss[0]).partSpec(map).build(), job);
      } else {
        System.err.println("Style of input: " + in + " is not right");
        System.exit(1);
      }
    }
    //解析用户的输出表字符串
    for (String out : outputs) {
      String[] ss = out.split("\\|");
      if (ss.length == 1) {
        OutputUtils.addTable(TableInfo.builder().tableName(ss[0]).build(), job);
      } else if (ss.length == 2) {
        LinkedHashMap<String, String> map = convertPartSpecToMap(ss[1]);
        OutputUtils.addTable(TableInfo.builder().tableName(ss[0]).partSpec(map).build(), job);
      } else if (ss.length == 3) {
        if (ss[1].isEmpty()) {
          LinkedHashMap<String, String> map = convertPartSpecToMap(ss[2]);
          OutputUtils.addTable(TableInfo.builder().tableName(ss[0]).partSpec(map).build(), job);
        } else {
          LinkedHashMap<String, String> map = convertPartSpecToMap(ss[1]);
          OutputUtils.addTable(TableInfo.builder().tableName(ss[0]).partSpec(map)
              .label(ss[2]).build(), job);
        }
      } else {
        System.err.println("Style of output: " + out + " is not right");
        System.exit(1);
      }
    }
    JobClient.runJob(job);
  }
}

多任务示例
更新时间：2018-12-12 10:54:39

编辑 ·
 · 我的收藏
本页目录
测试准备
测试步骤
预期结果
代码示例
测试准备
准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
准备好MultiJobs测试表和资源。
创建测试表。
试用
create table mr_empty (key string, value string);
create table mr_multijobs_out (value bigint);
添加测试资源。
试用
add table mr_multijobs_out as multijobs_res_table -f;
add jar data\resources\mapreduce-examples.jar -f;
测试步骤
在odpscmd中执行MultiJobs，如下所示：
试用
jar -resources mapreduce-examples.jar,multijobs_res_table -classpath data\resources\mapreduce-examples.jar
 com.aliyun.odps.mapred.open.example.MultiJobs mr_multijobs_out;
预期结果
作业成功结束后，输出表mr_multijobs_out中的内容，如下所示：
试用
+------------+
| value      |
+------------+
| 0          |
+------------+
代码示例
试用
    package com.aliyun.odps.mapred.open.example;
    import java.io.IOException;
    import java.util.Iterator;
    import com.aliyun.odps.data.Record;
    import com.aliyun.odps.data.TableInfo;
    import com.aliyun.odps.mapred.JobClient;
    import com.aliyun.odps.mapred.MapperBase;
    import com.aliyun.odps.mapred.RunningJob;
    import com.aliyun.odps.mapred.TaskContext;
    import com.aliyun.odps.mapred.conf.JobConf;
    import com.aliyun.odps.mapred.utils.InputUtils;
    import com.aliyun.odps.mapred.utils.OutputUtils;
    import com.aliyun.odps.mapred.utils.SchemaUtils;
    /**
     * MultiJobs
     *
     * Running multiple job
     *
     **/
    public class MultiJobs {
      public static class InitMapper extends MapperBase {
        @Override
        public void setup(TaskContext context) throws IOException {
          Record record = context.createOutputRecord();
          long v = context.getJobConf().getLong("multijobs.value", 2);
          record.set(0, v);
          context.write(record);
        }
      }
      public static class DecreaseMapper extends MapperBase {
        @Override
        public void cleanup(TaskContext context) throws IOException {
          //从JobConf中获取main函数中定义的变量值
          long expect = context.getJobConf().getLong("multijobs.expect.value", -1);
          long v = -1;
          int count = 0;
          //读取资源表里面的数据，这个表是上一个job的输出表
          Iterator<Record> iter = context.readResourceTable("multijobs_res_table");
          while (iter.hasNext()) {
            Record r = iter.next();
            v = (Long) r.get(0);
            if (expect != v) {
              throw new IOException("expect: " + expect + ", but: " + v);
            }
            count++;
          }
          if (count != 1) {
            throw new IOException("res_table should have 1 record, but: " + count);
          }
          Record record = context.createOutputRecord();
          v--;
          record.set(0, v);
          context.write(record);
          //设置counter，counter在作业成功结束后，可以在main函数中获取到
          context.getCounter("multijobs", "value").setValue(v);
        }
      }
      public static void main(String[] args) throws Exception {
        if (args.length != 1) {
          System.err.println("Usage: TestMultiJobs <table>");
          System.exit(1);
        }
        String tbl = args[0];
        long iterCount = 2;
        System.err.println("Start to run init job.");
        JobConf initJob = new JobConf();
        initJob.setLong("multijobs.value", iterCount);
        initJob.setMapperClass(InitMapper.class);
        InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), initJob);
        OutputUtils.addTable(TableInfo.builder().tableName(tbl).build(), initJob);
        initJob.setMapOutputKeySchema(SchemaUtils.fromString("key:string"));
        initJob.setMapOutputValueSchema(SchemaUtils.fromString("value:string"));
        //maponly作业需要显式设置reducer的数目为0
        initJob.setNumReduceTasks(0);
        JobClient.runJob(initJob);
        while (true) {
          System.err.println("Start to run iter job, count: " + iterCount);
          JobConf decJob = new JobConf();
          decJob.setLong("multijobs.expect.value", iterCount);
          decJob.setMapperClass(DecreaseMapper.class);
          InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), decJob);
          OutputUtils.addTable(TableInfo.builder().tableName(tbl).build(), decJob);
          //maponly作业需要显式设置reducer的数目为0
          decJob.setNumReduceTasks(0);
          RunningJob rJob = JobClient.runJob(decJob);
          iterCount--;
          //如果迭代次数已经达到，则退出循环
          if (rJob.getCounters().findCounter("multijobs", "value").getValue() == 0) {
            break;
          }
        }
        if (iterCount != 0) {
          throw new IOException("Job failed.");
        }
      }
    }

    二次排序示例
    更新时间：2018-12-12 10:55:21

    编辑 ·
     · 我的收藏
    本页目录
    测试准备
    测试步骤
    预期结果
    代码示例
    测试准备
    准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
    准备好SecondarySort的测试表和资源。
    创建测试表。
    试用
    create table ss_in(key bigint, value bigint);
    create table ss_out(key bigint, value bigint)
    添加测试资源。
    试用
    add jar data\resources\mapreduce-examples.jar -f;
    使用tunnel导入数据。
    试用
    tunnel upload data ss_in;
    导入ss_in表的数据文件data的内容，如下所示：
    试用
    1,2
    2,1
    1,1
    2,2
    测试步骤
    在odpscmd中执行SecondarySort，如下所示：
    试用
    jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
    com.aliyun.odps.mapred.open.example.SecondarySort ss_in ss_out;
    预期结果
    作业成功结束后，输出表ss_out中的内容，如下所示：
    试用
    +------------+------------+
    | key        | value      |
    +------------+------------+
    | 1          | 1          |
    | 1          | 2          |
    | 2          | 1          |
    | 2          | 2          |
    +------------+------------+
    代码示例
    试用
        package com.aliyun.odps.mapred.open.example;
        import java.io.IOException;
        import java.util.Iterator;
        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.TaskContext;
        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;
        /**
         *
         * This is an example ODPS Map/Reduce application. It reads the input table that
         * must contain two integers per record. The output is sorted by the first and
         * second number and grouped on the first number.
         *
         **/
        public class SecondarySort {
          /**
           * Read two integers from each line and generate a key, value pair as ((left,
           * right), right).
           **/
          public static class MapClass extends MapperBase {
            private Record key;
            private Record value;
            @Override
            public void setup(TaskContext context) throws IOException {
              key = context.createMapOutputKeyRecord();
              value = context.createMapOutputValueRecord();
            }
            @Override
            public void map(long recordNum, Record record, TaskContext context)
                throws IOException {
              long left = 0;
              long right = 0;
              if (record.getColumnCount() > 0) {
                left = (Long) record.get(0);
                if (record.getColumnCount() > 1) {
                  right = (Long) record.get(1);
                }
                key.set(new Object[] { (Long) left, (Long) right });
                value.set(new Object[] { (Long) right });
                context.write(key, value);
              }
            }
          }
          /**
           * A reducer class that just emits the sum of the input values.
           **/
          public static class ReduceClass extends ReducerBase {
            private Record result = null;
            @Override
            public void setup(TaskContext context) throws IOException {
              result = context.createOutputRecord();
            }
            @Override
            public void reduce(Record key, Iterator<Record> values, TaskContext context)
                throws IOException {
              result.set(0, key.get(0));
              while (values.hasNext()) {
                Record value = values.next();
                result.set(1, value.get(0));
                context.write(result);
              }
            }
          }
          public static void main(String[] args) throws Exception {
            if (args.length != 2) {
              System.err.println("Usage: secondarysrot <in> <out>");
              System.exit(2);
            }
            JobConf job = new JobConf();
            job.setMapperClass(MapClass.class);
            job.setReducerClass(ReduceClass.class);
            //将多列设置为Key
            //compare first and second parts of the pair
            job.setOutputKeySortColumns(new String[] { "i1", "i2" });
            //partition based on the first part of the pair
            job.setPartitionColumns(new String[] { "i1" });
            //grouping comparator based on the first part of the pair
            job.setOutputGroupingColumns(new String[] { "i1" });
            //the map output is LongPair, Long
            job.setMapOutputKeySchema(SchemaUtils.fromString("i1:bigint,i2:bigint"));
            job.setMapOutputValueSchema(SchemaUtils.fromString("i2x:bigint"));
            InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
            OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
            JobClient.runJob(job);
            System.exit(0);
          }
        }



        使用资源示例
        更新时间：2018-12-12 10:56:02

        编辑 ·
         · 我的收藏
        本页目录
        测试准备
        测试步骤
        预期结果
        代码示例
        测试准备
        准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
        准备好测试表和资源。
        创建测试表。
        试用
        create table mr_upload_src(key bigint, value string);
        添加测试资源。
        试用
        add jar data\resources\mapreduce-examples.jar -f;
        add file data\resources\import.txt -f;
        import.txt的数据内容，如下所示：
        试用
        1000,odps
        测试步骤
        在odpscmd中执行Upload，如下所示：
        试用
        jar -resources mapreduce-examples.jar,import.txt -classpath data\resources\mapreduce-examples.jar
        com.aliyun.odps.mapred.open.example.Upload import.txt mr_upload_src;
        预期结果
        作业成功结束后，输出表mr_upload_src中的内容，如下所示：
        试用
        +------------+------------+
        | key        | value      |
        +------------+------------+
        | 1000       | odps       |
        +------------+------------+
        代码示例
        试用
            package com.aliyun.odps.mapred.open.example;
            import java.io.BufferedInputStream;
            import java.io.FileNotFoundException;
            import java.io.IOException;
            import com.aliyun.odps.data.Record;
            import com.aliyun.odps.data.TableInfo;
            import com.aliyun.odps.mapred.JobClient;
            import com.aliyun.odps.mapred.MapperBase;
            import com.aliyun.odps.mapred.TaskContext;
            import com.aliyun.odps.mapred.conf.JobConf;
            import com.aliyun.odps.mapred.utils.InputUtils;
            import com.aliyun.odps.mapred.utils.OutputUtils;
            import com.aliyun.odps.mapred.utils.SchemaUtils;
            /**
             * Upload
             *
             * Import data from text file into table
             *
             **/
            public class Upload {
              public static class UploadMapper extends MapperBase {
                @Override
                public void setup(TaskContext context) throws IOException {
                  Record record = context.createOutputRecord();
                  StringBuilder importdata = new StringBuilder();
                  BufferedInputStream bufferedInput = null;
                  try {
                    byte[] buffer = new byte[1024];
                    int bytesRead = 0;
                    String filename = context.getJobConf().get("import.filename");
                    bufferedInput = context.readResourceFileAsStream(filename);
                    while ((bytesRead = bufferedInput.read(buffer)) != -1) {
                      String chunk = new String(buffer, 0, bytesRead);
                      importdata.append(chunk);
                    }
                    String lines[] = importdata.toString().split("\n");
                    for (int i = 0; i < lines.length; i++) {
                      String[] ss = lines[i].split(",");
                      record.set(0, Long.parseLong(ss[0].trim()));
                      record.set(1, ss[1].trim());
                      context.write(record);
                    }
                  } catch (FileNotFoundException ex) {
                    throw new IOException(ex);
                  } catch (IOException ex) {
                    throw new IOException(ex);
                  } finally {
                  }
                }
                @Override
                public void map(long recordNum, Record record, TaskContext context)
                    throws IOException {
                }
              }
              public static void main(String[] args) throws Exception {
                if (args.length != 2) {
                  System.err.println("Usage: Upload <import_txt> <out_table>");
                  System.exit(2);
                }
                JobConf job = new JobConf();
                job.setMapperClass(UploadMapper.class);
                //设置资源名字, 可以在map中通过jobconf获取到
                job.set("import.filename", args[0]);
                //maponly作业需要显式设置reducer的数目为0
                job.setNumReduceTasks(0);
                job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint"));
                job.setMapOutputValueSchema(SchemaUtils.fromString("value:string"));
                InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), job);
                OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
                JobClient.runJob(job);
              }
            }
        实际上，您可通过以下两种方式设置JobConf。

        通过SDK中JobConf的接口设置，本示例即是通过此方法实现。
        在Jar命令行中，通过–conf参数指定新的JobConf文件。

        使用Counter示例
        更新时间：2018-12-12 10:57:04

        编辑 ·
         · 我的收藏
        本页目录
        测试准备
        测试步骤
        预期结果
        代码示例
        测试准备
        准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
        准备好UserDefinedCounters测试表和资源。
        创建测试表。
        试用
        create table wc_in (key string, value string);
        create table wc_out(key string, cnt bigint);
        添加测试资源。
        试用
        add jar data\resources\mapreduce-examples.jar -f;
        使用tunnel导入数据。
        试用
        tunnel upload data wc_in;
        导入wc_in表的数据文件data的内容，如下所示：
        试用
        hello,odps
        测试步骤
        在odpscmd中执行UserDefinedCounters，如下所示：
        试用
        jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
        com.aliyun.odps.mapred.open.example.UserDefinedCounters wc_in wc_out
        预期结果
        作业成功结束后，可以看到Counters的输出，如下所示：
        试用
        Counters: 3
        com.aliyun.odps.mapred.open.example.UserDefinedCounters$MyCounter
        MAP_TASKS=1
        REDUCE_TASKS=1
        TOTAL_TASKS=2
        输出表wc_out中的内容，如下所示：
        试用
        +------------+------------+
        | key        | cnt        |
        +------------+------------+
        | hello      | 1          |
        | odps       | 1          |
        +------------+------------+
        代码示例
        试用
            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 {
              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);
                  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();
                  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);
                //在作业成功结束后，可以获取到job里面的自定义counter的值
                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);
              }
            }


            rep示例
            更新时间：2018-12-12 15:41:37

            编辑 ·
             · 我的收藏
            本页目录
            测试准备
            测试步骤
            预期结果
            代码示例
            测试准备
            准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
            准备好Grep测试表和资源。
            创建测试表。
            试用
            create table mr_src(key string, value string);
            create table mr_grep_tmp (key string, cnt bigint);
            create table mr_grep_out (key bigint, value string);
            添加测试资源。
            试用
            add jar data\resources\mapreduce-examples.jar -f;
            使用tunnel导入数据。
            试用
            tunnel upload data mr_src;
            导入mr_src表的数据文件data的内容，如下所示：
            试用
             hello,odps
             hello,world
            测试步骤
            在odpscmd中执行Grep，如下所示：
            试用
            jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
            com.aliyun.odps.mapred.open.example.Grep mr_src mr_grep_tmp mr_grep_out hello;
            预期结果
            作业成功结束后，输出表mr_grep_out中的内容，如下所示：
            试用
            +------------+------------+
            | key        | value      |
            +------------+------------+
            | 2          | hello      |
            +------------+------------+
            代码示例
            试用
                package com.aliyun.odps.mapred.open.example;
                import java.io.IOException;
                import java.util.Iterator;
                import java.util.regex.Matcher;
                import java.util.regex.Pattern;
                import com.aliyun.odps.data.Record;
                import com.aliyun.odps.data.TableInfo;
                import com.aliyun.odps.mapred.JobClient;
                import com.aliyun.odps.mapred.Mapper;
                import com.aliyun.odps.mapred.MapperBase;
                import com.aliyun.odps.mapred.ReducerBase;
                import com.aliyun.odps.mapred.RunningJob;
                import com.aliyun.odps.mapred.TaskContext;
                import com.aliyun.odps.mapred.conf.JobConf;
                import com.aliyun.odps.mapred.utils.InputUtils;
                import com.aliyun.odps.mapred.utils.OutputUtils;
                import com.aliyun.odps.mapred.utils.SchemaUtils;
                /**
                 *
                 * Extracts matching regexs from input files and counts them.
                 *
                 **/
                public class Grep {
                  /**
                   * RegexMapper
                   **/
                  public class RegexMapper extends MapperBase {
                    private Pattern pattern;
                    private int group;
                    private Record word;
                    private Record one;
                    @Override
                    public void setup(TaskContext context) throws IOException {
                      JobConf job = (JobConf) context.getJobConf();
                      pattern = Pattern.compile(job.get("mapred.mapper.regex"));
                      group = job.getInt("mapred.mapper.regex.group", 0);
                      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) {
                        String text = record.get(i).toString();
                        Matcher matcher = pattern.matcher(text);
                        while (matcher.find()) {
                          word.set(new Object[] { matcher.group(group) });
                          context.write(word, one);
                        }
                      }
                    }
                  }
                  /**
                   * LongSumReducer
                   **/
                  public class LongSumReducer extends ReducerBase {
                    private Record result = null;
                    @Override
                    public void setup(TaskContext context) throws IOException {
                      result = context.createOutputRecord();
                    }
                    @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);
                    }
                  }
                  /**
                   * A {@link Mapper} that swaps keys and values.
                   **/
                  public class InverseMapper extends MapperBase {
                    private Record word;
                    private Record count;
                    @Override
                    public void setup(TaskContext context) throws IOException {
                      word = context.createMapOutputValueRecord();
                      count = context.createMapOutputKeyRecord();
                    }
                    /**
                     * The inverse function. Input keys and values are swapped.
                     **/
                    @Override
                    public void map(long recordNum, Record record, TaskContext context) throws IOException {
                      word.set(new Object[] { record.get(0).toString() });
                      count.set(new Object[] { (Long) record.get(1) });
                      context.write(count, word);
                    }
                  }
                  /**
                   * IdentityReducer
                   **/
                  public class IdentityReducer extends ReducerBase {
                    private Record result = null;
                    @Override
                    public void setup(TaskContext context) throws IOException {
                      result = context.createOutputRecord();
                    }
                    /** Writes all keys and values directly to output. **/
                    @Override
                    public void reduce(Record key, Iterator<Record> values, TaskContext context) throws IOException {
                      result.set(0, key.get(0));
                      while (values.hasNext()) {
                        Record val = values.next();
                        result.set(1, val.get(0));
                        context.write(result);
                      }
                    }
                  }
                  public static void main(String[] args) throws Exception {
                    if (args.length < 4) {
                      System.err.println("Grep <inDir> <tmpDir> <outDir> <regex> [<group>]");
                      System.exit(2);
                    }
                    JobConf grepJob = new JobConf();
                    grepJob.setMapperClass(RegexMapper.class);
                    grepJob.setReducerClass(LongSumReducer.class);
                    grepJob.setMapOutputKeySchema(SchemaUtils.fromString("word:string"));
                    grepJob.setMapOutputValueSchema(SchemaUtils.fromString("count:bigint"));
                    InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), grepJob);
                    OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), grepJob);
                    //设置grepJob的grep的正则表达式
                    grepJob.set("mapred.mapper.regex", args[3]);
                    if (args.length == 5) {
                      grepJob.set("mapred.mapper.regex.group", args[4]);
                    }
                    @SuppressWarnings("unused")
                    RunningJob rjGrep = JobClient.runJob(grepJob);
                    //grepJob的输出作为sortJob的输入
                    JobConf sortJob = new JobConf();
                    sortJob.setMapperClass(InverseMapper.class);
                    sortJob.setReducerClass(IdentityReducer.class);
                    sortJob.setMapOutputKeySchema(SchemaUtils.fromString("count:bigint"));
                    sortJob.setMapOutputValueSchema(SchemaUtils.fromString("word:string"));
                    InputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), sortJob);
                    OutputUtils.addTable(TableInfo.builder().tableName(args[2]).build(), sortJob);
                    sortJob.setNumReduceTasks(1); // write a single file
                    sortJob.setOutputKeySortColumns(new String[] { "count" });
                    @SuppressWarnings("unused")
                    RunningJob rjSort = JobClient.runJob(sortJob);
                  }
                }




          Join示例
          更新时间：2018-12-12 10:58:20

          编辑 ·
           · 我的收藏
          本页目录
          测试准备
          测试步骤
          预期结果
          代码示例
          MaxCompute MapReduce框架自身并不支持Join逻辑，但您可以在自己的Map/Reduce函数中实现数据的Join，当然这需要您做一些额外的工作。

          假设需要Join两张表mr_Join_src1(key bigint, value string) 和mr_Join_src2(key bigint, value string)，输出表是mr_Join_out(key bigint, value1 string, value2 string)，其中value1是mr_Join_src1的value值，value2是 mr_Join_src2的value 值。

          测试准备
          准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
          准备好Join的测试表和资源。
          创建测试表。
          试用
          create table mr_Join_src1(key bigint, value string);
          create table mr_Join_src2(key bigint, value string);
          create table mr_Join_out(key bigint, value1 string, value2 string);
          添加测试资源。
          试用
          add jar data\resources\mapreduce-examples.jar -f;
          使用tunnel导入数据。
          试用
          tunnel upload data1 mr_Join_src1;
          tunnel upload data2 mr_Join_src2;
          导入mr_Join_src1数据的内容，如下所示：
          试用
           1,hello
           2,odps
          导入mr_Join_src2数据的内容，如下所示：
          试用
          1,odps
          3,hello
          4,odps
          测试步骤
          在odpscmd中执行Join，如下所示：
          试用
          jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
          com.aliyun.odps.mapred.open.example.Join mr_Join_src1 mr_Join_src2 mr_Join_out;
          预期结果
          作业成功结束后，输出表mr_Join_out中的内容，如下所示：
          试用
          +------------+------------+------------+
          | key        | value1     | value2     |
          +------------+------------+------------+
          |  1         | hello      |  odps      |
          +------------+------------+------------+
          代码示例
          试用
              package com.aliyun.odps.mapred.open.example;
              import java.io.IOException;
              import java.util.ArrayList;
              import java.util.Iterator;
              import java.util.List;
              import org.apache.commons.logging.Log;
              import org.apache.commons.logging.LogFactory;
              import com.aliyun.odps.data.Record;
              import com.aliyun.odps.data.TableInfo;
              import com.aliyun.odps.mapred.JobClient;
              import com.aliyun.odps.mapred.MapperBase;
              import com.aliyun.odps.mapred.ReducerBase;
              import com.aliyun.odps.mapred.conf.JobConf;
              import com.aliyun.odps.mapred.utils.InputUtils;
              import com.aliyun.odps.mapred.utils.OutputUtils;
              import com.aliyun.odps.mapred.utils.SchemaUtils;
              /**
               * Join, mr_Join_src1/mr_Join_src2(key bigint, value string), mr_Join_out(key
               * bigint, value1 string, value2 string)
               *
               */
              public class Join {
                public static final Log LOG = LogFactory.getLog(Join.class);
                public static class JoinMapper extends MapperBase {
                  private Record mapkey;
                  private Record mapvalue;
                  private long tag;
                  @Override
                  public void setup(TaskContext context) throws IOException {
                    mapkey = context.createMapOutputKeyRecord();
                    mapvalue = context.createMapOutputValueRecord();
                    tag = context.getInputTableInfo().getLabel().equals("left") ? 0 : 1;
                  }
                  @Override
                  public void map(long key, Record record, TaskContext context)
                      throws IOException {
                    mapkey.set(0, record.get(0));
                    mapkey.set(1, tag);
                    for (int i = 1; i < record.getColumnCount(); i++) {
                      mapvalue.set(i - 1, record.get(i));
                    }
                    context.write(mapkey, mapvalue);
                  }
                }
                public static class JoinReducer extends ReducerBase {
                  private Record result = null;
                  @Override
                  public void setup(TaskContext context) throws IOException {
                    result = context.createOutputRecord();
                  }
                  //reduce函数每次的输入会是key相同的所有record
                  @Override
                  public void reduce(Record key, Iterator<Record> values, TaskContext context)
                      throws IOException {
                    long k = key.getBigint(0);
                    List<Object[]> leftValues = new ArrayList<Object[]>();
                    //由于设置了outputKeySortColumn是key + tag组合，这样可以保证reduce函数的输入record中，left表的record数据在前面
                    while (values.hasNext()) {
                      Record value = values.next();
                      long tag = (Long) key.get(1);
                      //左表的数据会先缓存到内存中
                      if (tag == 0) {
                        leftValues.add(value.toArray().clone());
                      } else {
                        //碰到右表的数据，会与所有左表的数据进行join输出，此时左表的数据已经全部在内存里了
                   //这个实现只是一个功能展示，性能比较低，不建议用于实际生产
                        for (Object[] leftValue : leftValues) {
                          int index = 0;
                          result.set(index++, k);
                          for (int i = 0; i < leftValue.length; i++) {
                            result.set(index++, leftValue[i]);
                          }
                          for (int i = 0; i < value.getColumnCount(); i++) {
                            result.set(index++, value.get(i));
                          }
                          context.write(result);
                        }
                      }
                    }
                  }
                }
                public static void main(String[] args) throws Exception {
                  if (args.length != 3) {
                    System.err.println("Usage: Join <input table1> <input table2> <out>");
                    System.exit(2);
                  }
                  JobConf job = new JobConf();
                  job.setMapperClass(JoinMapper.class);
                  job.setReducerClass(JoinReducer.class);
                  job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint,tag:bigint"));
                  job.setMapOutputValueSchema(SchemaUtils.fromString("value:string"));
                  job.setPartitionColumns(new String[]{"key"});
                  job.setOutputKeySortColumns(new String[]{"key", "tag"});
                  job.setOutputGroupingColumns(new String[]{"key"});
                  job.setNumReduceTasks(1);
                  InputUtils.addTable(TableInfo.builder().tableName(args[0]).label("left").build(), job);
                  InputUtils.addTable(TableInfo.builder().tableName(args[1]).label("right").build(), job);
                  OutputUtils.addTable(TableInfo.builder().tableName(args[2]).build(), job);
                  JobClient.runJob(job);
                }
              }

Sleep示例
更新时间：2018-12-12 10:59:08

编辑 ·
 · 我的收藏
本页目录
测试准备
测试步骤
预期结果
代码示例
测试准备
准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
准备好SleepJob的测试资源。
试用
add jar data\resources\mapreduce-examples.jar -f;
测试步骤
在odpscmd中执行Sleep，如下所示：
试用
  jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
  com.aliyun.odps.mapred.open.example.Sleep 10;
  jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
  com.aliyun.odps.mapred.open.example.Sleep 100;
预期结果
作业成功结束后，对比不同Sleep时长的运行时间，可以看到效果。

代码示例
试用
package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.conf.JobConf;
public class Sleep {
  private static final String SLEEP_SECS = "sleep.secs";
  public static class MapperClass extends MapperBase {
    //由于没有输入数据，map函数不会被执行，相关逻辑只能写在setup中
    @Override
    public void setup(TaskContext context) throws IOException {
      try {
        //获取jobconf中设置的sleep秒数，进行sleep
        Thread.sleep(context.getJobConf().getInt(SLEEP_SECS, 1) * 1000);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }
  public static void main(String[] args) throws Exception {
    if (args.length != 1) {
      System.err.println("Usage: Sleep <sleep_secs>");
      System.exit(-1);
    }
    JobConf job = new JobConf();
    job.setMapperClass(MapperClass.class);
    //这个实例也是一个MapOnly的，所以需要设置reducer个数为0
    job.setNumReduceTasks(0);
    //由于没有输入表，mapper的个数需要由用户明确进行指定
    job.setNumMapTasks(1);
    job.set(SLEEP_SECS, args[0]);
    JobClient.runJob(job);
  }
}

Unique示例
更新时间：2018-12-12 11:00:11

编辑 ·
 · 我的收藏
本页目录
测试准备
测试步骤
预期结果
代码示例
测试准备
准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
准备好Unique的测试表和资源。
创建测试表。
试用
create table ss_in(key bigint, value bigint);
create table ss_out(key bigint, value bigint);
添加测试资源。
试用
add jar data\resources\mapreduce-examples.jar -f;
使用tunnel导入数据。
试用
tunnel upload data ss_in;
导入ss_in表的数据文件data的内容，如下所示：
试用
 1,1
 1,1
 2,2
 2,2
测试步骤
在odpscmd中执行Unique，如下所示：
试用
jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.Unique ss_in ss_out key;
预期结果
作业成功结束后，输出表ss_out中的内容，如下所示：
试用
+------------+------------+
| key        | value      |
+------------+------------+
| 1          |   1        |
| 2          |   2        |
+------------+------------+
代码示例
试用
    package com.aliyun.odps.mapred.open.example;
    import java.io.IOException;
    import java.util.Iterator;
    import com.aliyun.odps.data.Record;
    import com.aliyun.odps.data.TableInfo;
    import com.aliyun.odps.mapred.JobClient;
    import com.aliyun.odps.mapred.MapperBase;
    import com.aliyun.odps.mapred.ReducerBase;
    import com.aliyun.odps.mapred.TaskContext;
    import com.aliyun.odps.mapred.conf.JobConf;
    import com.aliyun.odps.mapred.utils.InputUtils;
    import com.aliyun.odps.mapred.utils.OutputUtils;
    import com.aliyun.odps.mapred.utils.SchemaUtils;
    /**
     * Unique Remove duplicate words
     *
     **/
    public class Unique {
      public static class OutputSchemaMapper extends MapperBase {
        private Record key;
        private Record value;
        @Override
        public void setup(TaskContext context) throws IOException {
          key = context.createMapOutputKeyRecord();
          value = context.createMapOutputValueRecord();
        }
        @Override
        public void map(long recordNum, Record record, TaskContext context)
            throws IOException {
          long left = 0;
          long right = 0;
          if (record.getColumnCount() > 0) {
            left = (Long) record.get(0);
            if (record.getColumnCount() > 1) {
              right = (Long) record.get(1);
            }
            key.set(new Object[] { (Long) left, (Long) right });
            value.set(new Object[] { (Long) left, (Long) right });
            context.write(key, value);
          }
        }
      }
      public static class OutputSchemaReducer extends ReducerBase {
        private Record result = null;
        @Override
        public void setup(TaskContext context) throws IOException {
          result = context.createOutputRecord();
        }
        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
            throws IOException {
          result.set(0, key.get(0));
          while (values.hasNext()) {
            Record value = values.next();
            result.set(1, value.get(1));
          }
          context.write(result);
        }
      }
      public static void main(String[] args) throws Exception {
        if (args.length > 3 || args.length < 2) {
          System.err.println("Usage: unique <in> <out> [key|value|all]");
          System.exit(2);
        }
        String ops = "all";
        if (args.length == 3) {
          ops = args[2];
        }
        // reduce的输入分组是由setOutputGroupingColumns的设置来决定的，这个参数如果不设置
        // 默认就是MapOutputKeySchema
        // Key Unique
        if (ops.equals("key")) {
          JobConf job = new JobConf();
          job.setMapperClass(OutputSchemaMapper.class);
          job.setReducerClass(OutputSchemaReducer.class);
          job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint,value:bigint"));
          job.setMapOutputValueSchema(SchemaUtils.fromString("key:bigint,value:bigint"));
          job.setPartitionColumns(new String[] { "key" });
          job.setOutputKeySortColumns(new String[] { "key", "value" });
          job.setOutputGroupingColumns(new String[] { "key" });
          job.set("tablename2", args[1]);
          job.setNumReduceTasks(1);
          job.setInt("table.counter", 0);
          InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
          OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
          JobClient.runJob(job);
        }
        // Key&Value Unique
        if (ops.equals("all")) {
          JobConf job = new JobConf();
          job.setMapperClass(OutputSchemaMapper.class);
          job.setReducerClass(OutputSchemaReducer.class);
          job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint,value:bigint"));
          job.setMapOutputValueSchema(SchemaUtils.fromString("key:bigint,value:bigint"));
          job.setPartitionColumns(new String[] { "key" });
          job.setOutputKeySortColumns(new String[] { "key", "value" });
          job.setOutputGroupingColumns(new String[] { "key", "value" });
          job.set("tablename2", args[1]);
          job.setNumReduceTasks(1);
          job.setInt("table.counter", 0);
          InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
          OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
          JobClient.runJob(job);
        }
        // Value Unique
        if (ops.equals("value")) {
          JobConf job = new JobConf();
          job.setMapperClass(OutputSchemaMapper.class);
          job.setReducerClass(OutputSchemaReducer.class);
          job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint,value:bigint"));
          job.setMapOutputValueSchema(SchemaUtils.fromString("key:bigint,value:bigint"));
          job.setPartitionColumns(new String[] { "value" });
          job.setOutputKeySortColumns(new String[] { "value" });
          job.setOutputGroupingColumns(new String[] { "value" });
          job.set("tablename2", args[1]);
          job.setNumReduceTasks(1);
          job.setInt("table.counter", 0);
          InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
          OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
          JobClient.runJob(job);
        }
      }
    }

    Sort示例
    更新时间：2018-12-12 11:00:54

    编辑 ·
     · 我的收藏
    本页目录
    测试准备
    测试步骤
    预期结果
    代码示例
    测试准备
    准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
    准备好Sort的测试表和资源。
    创建测试表。
    试用
    create table ss_in(key bigint, value bigint);
    create table ss_out(key bigint, value bigint);
    添加测试资源。
    试用
    add jar data\resources\mapreduce-examples.jar -f;
    使用tunnel导入数据。
    试用
    tunnel upload data ss_in;
    导入ss_in表的数据文件data的内容，如下所示：
    试用
     2,1
     1,1
     3,1
    测试步骤
    在odpscmd中执行Sort，如下所示：
    试用
    jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
    com.aliyun.odps.mapred.open.example.Sort ss_in ss_out;
    预期结果
    作业成功结束后，输出表ss_out中的内容，如下所示：
    试用
    +------------+------------+
    | key        | value      |
    +------------+------------+
    | 1          | 1          |
    | 2          | 1          |
    | 3          | 1          |
    +------------+------------+
    代码示例
    试用
        package com.aliyun.odps.mapred.open.example;
        import java.io.IOException;
        import java.util.Date;
        import com.aliyun.odps.data.Record;
        import com.aliyun.odps.data.TableInfo;
        import com.aliyun.odps.mapred.JobClient;
        import com.aliyun.odps.mapred.MapperBase;
        import com.aliyun.odps.mapred.TaskContext;
        import com.aliyun.odps.mapred.conf.JobConf;
        import com.aliyun.odps.mapred.example.lib.IdentityReducer;
        import com.aliyun.odps.mapred.utils.InputUtils;
        import com.aliyun.odps.mapred.utils.OutputUtils;
        import com.aliyun.odps.mapred.utils.SchemaUtils;
        /**
         * This is the trivial map/reduce program that does absolutely nothing other
         * than use the framework to fragment and sort the input values.
         *
         **/
        public class Sort {
          static int printUsage() {
            System.out.println("sort <input> <output>");
            return -1;
          }
          /**
           * Implements the identity function, mapping record's first two columns to
           * outputs.
           **/
          public static class IdentityMapper extends MapperBase {
            private Record key;
            private Record value;
            @Override
            public void setup(TaskContext context) throws IOException {
              key = context.createMapOutputKeyRecord();
              value = context.createMapOutputValueRecord();
            }
            @Override
            public void map(long recordNum, Record record, TaskContext context)
                throws IOException {
              key.set(new Object[] { (Long) record.get(0) });
              value.set(new Object[] { (Long) record.get(1) });
              context.write(key, value);
            }
          }
          /**
           * The main driver for sort program. Invoke this method to submit the
           * map/reduce job.
           *
           * @throws IOException
           *           When there is communication problems with the job tracker.
           **/
          public static void main(String[] args) throws Exception {
            JobConf jobConf = new JobConf();
            jobConf.setMapperClass(IdentityMapper.class);
            jobConf.setReducerClass(IdentityReducer.class);
            //为了全局有序，这里设置了reducer的个数为1，所有的数据都会集中到一个reducer上面
            //只能用于小数据量，大数据量需要考虑其他的方式，比如TeraSort
            jobConf.setNumReduceTasks(1);
            jobConf.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint"));
            jobConf.setMapOutputValueSchema(SchemaUtils.fromString("value:bigint"));
            InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), jobConf);
            OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), jobConf);
            Date startTime = new Date();
            System.out.println("Job started: " + startTime);
            JobClient.runJob(jobConf);
            Date end_time = new Date();
            System.out.println("Job ended: " + end_time);
            System.out.println("The job took "
                + (end_time.getTime() - startTime.getTime()) / 1000 + " seconds.");
          }
        }

分区表输入示例
更新时间：2018-12-12 11:01:33

编辑 ·
 · 我的收藏
本文为您介绍两个把Partition作为输入输出的示例，仅供参考。

示例一：
试用
     public static void main(String[] args) throws Exception {
     JobConf job = new JobConf();
     ...
     LinkedHashMap<String, String> input = new LinkedHashMap<String, String>();
     input.put("pt", "123456");
     InputUtils.addTable(TableInfo.builder().tableName("input_table").partSpec(input).build(), job);
     LinkedHashMap<String, String> output = new LinkedHashMap<String, String>();
     output.put("ds", "654321");
     OutputUtils.addTable(TableInfo.builder().tableName("output_table").partSpec(output).build(), job);
     JobClient.runJob(job);
    }
示例二：
试用
    package com.aliyun.odps.mapred.open.example;
    ...
      public static void main(String[] args) throws Exception {
        if (args.length != 2) {
          System.err.println("Usage: WordCount <in_table> <out_table>");
          System.exit(2);
        }
        JobConf job = new JobConf();
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(SumCombiner.class);
        job.setReducerClass(SumReducer.class);
        job.setMapOutputKeySchema(SchemaUtils.fromString("word:string"));
        job.setMapOutputValueSchema(SchemaUtils.fromString("count:bigint"));
        Account account = new AliyunAccount("my_access_id", "my_access_key");
        Odps odps = new Odps(account);
        odps.setEndpoint("odps_endpoint_url");
        odps.setDefaultProject("my_project");
        Table table = odps.tables().get(tblname);
        TableInfoBuilder builder = TableInfo.builder().tableName(tblname);
        for (Partition p : table.getPartitions()) {
          if (applicable(p)) {
            LinkedHashMap<String, String> partSpec = new LinkedHashMap<String, String>();
            for (String key : p.getPartitionSpec().keys()) {
              partSpec.put(key, p.getPartitionSpec().get(key));
            }
            InputUtils.addTable(builder.partSpec(partSpec).build(), conf);
          }
        }
        OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
        JobClient.runJob(job);
      }
说明
这是一段使用MaxCompute SDK和MapReduce SDK组合实现MapReduce任务读取范围Partitoin的示例。
此段代码不能够编译执行，仅给出了main函数的示例。
示例中applicable函数是用户逻辑，用来决定该Partition是否符合作为该MapReduce作业的输入。


Pipeline示例
更新时间：2018-12-12 11:02:11

编辑 ·
 · 我的收藏
本页目录
测试准备
测试步骤
预期结果
代码示例
测试准备
准备好测试程序的Jar包，假设名字为mapreduce-examples.jar，本地存放路径为data\resources。
准备好Pipeline的测试表和资源。
创建测试表。
试用
create table wc_in (key string, value string);
create table wc_out(key string, cnt bigint);
添加测试资源。
试用
add jar data\resources\mapreduce-examples.jar -f;
使用tunnel导入数据。
试用
tunnel upload data wc_in;
导入wc_in表的数据文件data的内容，如下所示：
试用
hello,odps
测试步骤
在odpscmd中执行WordCountPipeline，如下所示：
试用
jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.WordCountPipeline wc_in wc_out;
预期结果
作业成功结束后，输出表wc_out中的内容，如下所示：
试用
+------------+------------+
| key        | cnt        |
+------------+------------+
| hello      | 1          |
| odps       | 1          |
+------------+------------+
代码示例
试用
    package com.aliyun.odps.mapred.open.example;
    import java.io.IOException;
    import java.util.Iterator;
    import com.aliyun.odps.Column;
    import com.aliyun.odps.OdpsException;
    import com.aliyun.odps.OdpsType;
    import com.aliyun.odps.data.Record;
    import com.aliyun.odps.data.TableInfo;
    import com.aliyun.odps.mapred.Job;
    import com.aliyun.odps.mapred.MapperBase;
    import com.aliyun.odps.mapred.ReducerBase;
    import com.aliyun.odps.pipeline.Pipeline;
    public class WordCountPipelineTest {
      public static class TokenizerMapper extends MapperBase {
        Record word;
        Record one;
        @Override
        public void setup(TaskContext context) throws IOException {
          word = context.createMapOutputKeyRecord();
          one = context.createMapOutputValueRecord();
          one.setBigint(0, 1L);
        }
        @Override
        public void map(long recordNum, Record record, TaskContext context)
            throws IOException {
          for (int i = 0; i < record.getColumnCount(); i++) {
            String[] words = record.get(i).toString().split("\\s+");
            for (String w : words) {
              word.setString(0, w);
              context.write(word, one);
            }
          }
        }
      }
      public static class SumReducer extends ReducerBase {
        private Record value;
        @Override
        public void setup(TaskContext context) throws IOException {
          value = context.createOutputValueRecord();
        }
        @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);
          }
          value.set(0, count);
          context.write(key, value);
        }
      }
      public static class IdentityReducer extends ReducerBase {
        private Record result;
        @Override
        public void setup(TaskContext context) throws IOException {
          result = context.createOutputRecord();
        }
        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
            throws IOException {
          while (values.hasNext()) {
            result.set(0, key.get(0));
            result.set(1, values.next().get(0));
            context.write(result);
          }
        }
      }
      public static void main(String[] args) throws OdpsException {
        if (args.length != 2) {
          System.err.println("Usage: WordCountPipeline <in_table> <out_table>");
          System.exit(2);
        }
        Job job = new Job();
        /***
         * 构造Pipeline的过程中，如果不指定Mapper的OutputKeySortColumns，PartitionColumns，OutputGroupingColumns，
         * 框架会默认使用其OutputKey作为此三者的默认配置
         ***/
        Pipeline pipeline = Pipeline.builder()
            .addMapper(TokenizerMapper.class)
            .setOutputKeySchema(
                    new Column[] { new Column("word", OdpsType.STRING) })
            .setOutputValueSchema(
                    new Column[] { new Column("count", OdpsType.BIGINT) })
            .setOutputKeySortColumns(new String[] { "word" })
            .setPartitionColumns(new String[] { "word" })
            .setOutputGroupingColumns(new String[] { "word" })
            .addReducer(SumReducer.class)
            .setOutputKeySchema(
                    new Column[] { new Column("word", OdpsType.STRING) })
            .setOutputValueSchema(
                    new Column[] { new Column("count", OdpsType.BIGINT)})
            .addReducer(IdentityReducer.class).createPipeline();
        //将pipeline的设置到jobconf中，如果需要设置combiner，是通过jobconf来设置
        job.setPipeline(pipeline);
        //设置输入输出表
        job.addInput(TableInfo.builder().tableName(args[0]).build());
        job.addOutput(TableInfo.builder().tableName(args[1]).build());
        //作业提交并等待结束
        job.submit();
        job.waitForCompletion();
        System.exit(job.isSuccessful() == true ? 0 : 1);
      }
    }