643      阿裏雲

join示例__示例程序_MapReduce_大數據計算服務-阿裏雲

ODPS 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值。

測試準備

（1）準備好測試程序jar包，假設名字為mapreduce-examples.jar;

（2）準備好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 mapreduce-examples.jar -f;
  

（3）使用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 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();
        }

        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
            throws IOException {
          long k = key.getBigint(0);
          List<Object[]> leftValues = new ArrayList<Object[]>();

          while (values.hasNext()) {
            Record value = values.next();
            long tag = (Long) key.get(1);

            if (tag == 0) {
              leftValues.add(value.toArray().clone());
            } else {
              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);
      }

    }

最後更新：2016-05-06 10:43:08

  上一篇： grep示例__示例程序_MapReduce_大數據計算服務-阿裏雲

  下一篇： sleep示例__示例程序_MapReduce_大數據計算服務-阿裏雲