AI 助理
备案控制台
开发者社区 问答 正文

MaxCompute用户指南：MapReduce：示例程序：Unique示例



测试准备


  1. 准备好测试程序的 Jar 包，假设名字为 mapreduce-examples.jar，本地存放路径为 data\resources。

  2. 准备好 Unique 的测试表和资源。

    • 创建测试表。create table ss_in(key bigint, value bigint);
    • create table ss_out(key bigint, value bigint);

  • 添加测试资源。
    1. add jar data\resources\mapreduce-examples.jar -f;

    使用 tunnel 导入数据。
    1. tunnel upload data ss_in;

    导入 ss_in 表的数据文件 data 的内容，如下所示：
    1. 1,1
    2. 1,1
    3. 2,2
    4. 2,2


    测试步骤


    在 odpscmd 中执行 Unique，如下所示：
    1. jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
    2. com.aliyun.odps.mapred.open.example.Unique ss_in ss_out key;


    预期结果


    作业成功结束后，输出表 ss_out 中的内容，如下所示：
    1. +------------+------------+
    2. | key        | value      |
    3. +------------+------------+
    4. | 1          |   1        |
    5. | 2          |   2        |
    6. +------------+------------+


    代码示例

    1.     package com.aliyun.odps.mapred.open.example;
    3.     import java.io.IOException;
    4.     import java.util.Iterator;
    6.     import com.aliyun.odps.data.Record;
    7.     import com.aliyun.odps.data.TableInfo;
    8.     import com.aliyun.odps.mapred.JobClient;
    9.     import com.aliyun.odps.mapred.MapperBase;
    10.     import com.aliyun.odps.mapred.ReducerBase;
    11.     import com.aliyun.odps.mapred.TaskContext;
    12.     import com.aliyun.odps.mapred.conf.JobConf;
    13.     import com.aliyun.odps.mapred.utils.InputUtils;
    14.     import com.aliyun.odps.mapred.utils.OutputUtils;
    15.     import com.aliyun.odps.mapred.utils.SchemaUtils;
    17.     /**
    18.      * Unique Remove duplicate words
    19.      *
    20.      **/
    21.     public class Unique {
    23.       public static class OutputSchemaMapper extends MapperBase {
    24.         private Record key;
    25.         private Record value;
    27.         @Override
    28.         public void setup(TaskContext context) throws IOException {
    29.           key = context.createMapOutputKeyRecord();
    30.           value = context.createMapOutputValueRecord();
    31.         }
    33.         @Override
    34.         public void map(long recordNum, Record record, TaskContext context)
    35.             throws IOException {
    36.           long left = 0;
    37.           long right = 0;
    39.           if (record.getColumnCount() > 0) {
    40.             left = (Long) record.get(0);
    41.             if (record.getColumnCount() > 1) {
    42.               right = (Long) record.get(1);
    43.             }
    45.             key.set(new Object[] { (Long) left, (Long) right });
    46.             value.set(new Object[] { (Long) left, (Long) right });
    48.             context.write(key, value);
    49.           }
    50.         }
    51.       }
    53.       public static class OutputSchemaReducer extends ReducerBase {
    54.         private Record result = null;
    56.         @Override
    57.         public void setup(TaskContext context) throws IOException {
    58.           result = context.createOutputRecord();
    59.         }
    61.         @Override
    62.         public void reduce(Record key, Iterator<Record> values, TaskContext context)
    63.             throws IOException {
    64.           result.set(0, key.get(0));
    65.           while (values.hasNext()) {
    66.             Record value = values.next();
    67.             result.set(1, value.get(1));
    68.           }
    69.           context.write(result);
    70.         }
    71.       }
    73.       public static void main(String[] args) throws Exception {
    74.         if (args.length > 3 || args.length < 2) {
    75.           System.err.println("Usage: unique <in> <out> [key|value|all]");
    76.           System.exit(2);
    77.         }
    79.         String ops = "all";
    80.         if (args.length == 3) {
    81.           ops = args[2];
    82.         }
    84.         // Key Unique
    85.         if (ops.equals("key")) {
    86.           JobConf job = new JobConf();
    88.           job.setMapperClass(OutputSchemaMapper.class);
    89.           job.setReducerClass(OutputSchemaReducer.class);
    91.           job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint,value:bigint"));
    92.           job.setMapOutputValueSchema(SchemaUtils.fromString("key:bigint,value:bigint"));
    94.           job.setPartitionColumns(new String[] { "key" });
    95.           job.setOutputKeySortColumns(new String[] { "key", "value" });
    96.           job.setOutputGroupingColumns(new String[] { "key" });
    98.           job.set("tablename2", args[1]);
    100.           job.setNumReduceTasks(1);
    101.           job.setInt("table.counter", 0);
    103.           InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
    104.           OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
    106.           JobClient.runJob(job);
    107.         }
    109.         // Key&Value Unique
    110.         if (ops.equals("all")) {
    111.           JobConf job = new JobConf();
    113.           job.setMapperClass(OutputSchemaMapper.class);
    114.           job.setReducerClass(OutputSchemaReducer.class);
    116.           job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint,value:bigint"));
    117.           job.setMapOutputValueSchema(SchemaUtils.fromString("key:bigint,value:bigint"));
    119.           job.setPartitionColumns(new String[] { "key" });
    120.           job.setOutputKeySortColumns(new String[] { "key", "value" });
    121.           job.setOutputGroupingColumns(new String[] { "key", "value" });
    123.           job.set("tablename2", args[1]);
    125.           job.setNumReduceTasks(1);
    126.           job.setInt("table.counter", 0);
    128.           InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
    129.           OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
    131.           JobClient.runJob(job);
    132.         }
    134.         // Value Unique
    135.         if (ops.equals("value")) {
    136.           JobConf job = new JobConf();
    138.           job.setMapperClass(OutputSchemaMapper.class);
    139.           job.setReducerClass(OutputSchemaReducer.class);
    141.           job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint,value:bigint"));
    142.           job.setMapOutputValueSchema(SchemaUtils.fromString("key:bigint,value:bigint"));
    144.           job.setPartitionColumns(new String[] { "value" });
    145.           job.setOutputKeySortColumns(new String[] { "value" });
    146.           job.setOutputGroupingColumns(new String[] { "value" });
    148.           job.set("tablename2", args[1]);
    150.           job.setNumReduceTasks(1);
    151.           job.setInt("table.counter", 0);
    153.           InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
    154.           OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
    156.           JobClient.runJob(job);
    157.         }
    159.       }
    161.     }

    • 展开
    收起
    行者武松 2017-10-23 17:47:52 2076 分享 版权

    版权声明：本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

    0 条回答
    写回答
    取消 提交回答
    问答分类：
    分布式计算 Java MaxCompute 云原生大数据计算服务 MaxCompute
    问答标签：
    云原生大数据计算服务 MaxCompute示例 云原生大数据计算服务 MaxCompute程序 云原生大数据计算服务 MaxCompute mapreduce maxcompute mapreduce mapreduce示例
    问答地址：
    开发者社区 > 大数据 > 问答
    相关问答
    通过ECS自建的大数据平台迁移到E-MapReduce是否需要停机
    106
    1
    0
    MaxCompute提交mapreduce作业处理oss上的数据报错Exception in ...
    128
    1
    0
    E-MapReduce集群弹性扩容后其gateway集群提交大数据作业失败
    83
    0
    0
    EMR（E-MapReduce）是阿里巴巴飞天大数据平台的哪个产品，它的主要功能是什么？
    237
    1
    0
    为从MaxCompute提取数据并写入Hologre，上面的SQL语句如何编写，可以提供一个示例吗？
    61
    0
    0
    大数据计算MaxCompute是可以用hdfs和hive和mapreduce的吗？
    108
    0
    0
    mapreduce任务能否往MaxCompute分区表写数据
    82
    1
    0
    MapReduce任务如何调用MaxCompute的API
    80
    1
    0
    原生Apache Hadoop使用MapReduce整合odps读取MaxCompute表数据...
    95
    0
    0
    通过mapreduce任务写数据到MaxCompute中如何支持追加操作
    56
    1
    0