开发者社区 问答 正文

MaxCompute用户指南:MapReduce:示例程序:MapOnly示例



对于 MapOnly 的作业,Map 直接将 < Key, Value > 信息输出到 MaxCompute 的表中,您只需要指定输出表即可,不需指定 Map 输出的 Key/Value 元信息。

测试准备


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

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

    • 创建测试表create table wc_in (key string, value string);
    • create table wc_out(key string, cnt bigint);

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

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

    导入 wc_in 表的数据文件 data 的内容,如下所示:
    1. hello,odps
    2. hello,odps


    测试步骤


    在 odpscmd 中执行 MapOnly,如下所示:
    1. jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
    2. com.aliyun.odps.mapred.open.example.MapOnly wc_in wc_out map


    预期结果


    作业成功结束后,输出表 wc_out 中的内容,如下所示:
    1. +------------+------------+
    2. | key        | cnt        |
    3. +------------+------------+
    4. | hello      | 1          |
    5. | hello      | 1          |
    6. +------------+------------+


    代码示例

    1.     package com.aliyun.odps.mapred.open.example;
    2.     import java.io.IOException;
    3.     import com.aliyun.odps.data.Record;
    4.     import com.aliyun.odps.mapred.JobClient;
    5.     import com.aliyun.odps.mapred.MapperBase;
    6.     import com.aliyun.odps.mapred.conf.JobConf;
    7.     import com.aliyun.odps.mapred.utils.SchemaUtils;
    8.     import com.aliyun.odps.mapred.utils.InputUtils;
    9.     import com.aliyun.odps.mapred.utils.OutputUtils;
    10.     import com.aliyun.odps.data.TableInfo;
    11.     public class MapOnly {
    12.       public static class MapperClass extends MapperBase {
    13.         @Override
    14.         public void setup(TaskContext context) throws IOException {
    15.           boolean is = context.getJobConf().getBoolean("option.mapper.setup", false);
    16.           if (is) {
    17.             Record result = context.createOutputRecord();
    18.             result.set(0, "setup");
    19.             result.set(1, 1L);
    20.             context.write(result);
    21.           }
    22.         }
    23.         @Override
    24.         public void map(long key, Record record, TaskContext context) throws IOException {
    25.           boolean is = context.getJobConf().getBoolean("option.mapper.map", false);
    26.           if (is) {
    27.             Record result = context.createOutputRecord();
    28.             result.set(0, record.get(0));
    29.             result.set(1, 1L);
    30.             context.write(result);
    31.           }
    32.         }
    33.         @Override
    34.         public void cleanup(TaskContext context) throws IOException {
    35.           boolean is = context.getJobConf().getBoolean("option.mapper.cleanup", false);
    36.           if (is) {
    37.             Record result = context.createOutputRecord();
    38.             result.set(0, "cleanup");
    39.             result.set(1, 1L);
    40.             context.write(result);
    41.           }
    42.         }
    43.       }
    44.       public static void main(String[] args) throws Exception {
    45.         if (args.length != 2 && args.length != 3) {
    46.           System.err.println("Usage: OnlyMapper <in_table> <out_table> [setup|map|cleanup]");
    47.           System.exit(2);
    48.         }
    49.         JobConf job = new JobConf();
    50.         job.setMapperClass(MapperClass.class);
    51.         job.setNumReduceTasks(0);
    52.         InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
    53.         OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
    54.         if (args.length == 3) {
    55.           String options = new String(args[2]);
    56.           if (options.contains("setup")) {
    57.             job.setBoolean("option.mapper.setup", true);
    58.           }
    59.           if (options.contains("map")) {
    60.             job.setBoolean("option.mapper.map", true);
    61.           }
    62.           if (options.contains("cleanup")) {
    63.             job.setBoolean("option.mapper.cleanup", true);
    64.           }
    65.         }
    66.         JobClient.runJob(job);
    67.       }
    68.     }
  • 展开
    收起
    行者武松 2017-10-23 17:44:01 2107 分享 版权
    0 条回答
    写回答
    取消 提交回答