MultipleInOut example - MaxCompute - Alibaba Cloud Dokümantasyon Merkezi

Ön koşullar

Başlamadan önce aşağıdakilere sahip olduğunuzdan emin olun:

Başlangıç bölümünde açıklanan ortam kurulumunu tamamladınız
MaxCompute istemci kurulumunuzun bin\data\resources dizininde bulunan mapreduce-examples.jar dosyasına sahipsiniz

Test tablolarını ve kaynaklarını ayarlayın

Tabloları oluşturun

İki giriş tablosu ve iki çıktı tablosu oluşturmak için MaxCompute istemcinizde aşağıdaki ifadeleri çalıştırın. mr_multiinout_out2 tablosu bölümlenmiştir; işi çalıştırmadan önce bu tabloya iki bölüm ekleyin.

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

JAR kaynağını ekleyin

-- When adding the JAR package for the first time, you can ignore the -f flag.
add jar data\resources\mapreduce-examples.jar -f;

Test verilerini yükleyin

Tunnel'ı kullanarak MaxCompute istemcinizin bin dizinindeki data1.txt ve data2.txt dosyalarını iki giriş tablosuna yükleyin.

tunnel upload data1.txt wc_in1;
tunnel upload data2.txt wc_in2;

Yükleme işleminden sonra tablolar aşağıdaki verileri içerir:

wc_in1:

hello,odps

wc_in2:

hello,world

İşi çalıştırın

MaxCompute istemcinizde aşağıdaki komutu çalıştırın:

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;

Komut iki konumsal bağımsız değişken alır:

Bağımsız Değişken	Biçim	Örnek
Girişler	Virgülle ayrılmış tablo adları	`wc_in1,wc_in2`
Çıktılar	Her biri `table_name\|partition_spec\|label` biçiminde virgülle ayrılmış girdiler	`mr_multiinout_out2\|a=1/b=1\|out1`

Çıktılar için partition_spec ve label isteğe bağlıdır. label belirtmezseniz çıktı varsayılan (etiketsiz) hedefe yönlendirilir.

Beklenen sonuçlar

Reducer, her anahtarı anahtarın toplam sayısının 3'e bölümünden kalan değerine göre belirli bir çıktı tablosuna yönlendirir:

Koşul	Hedef	Çıktı tablosu
`count % 3 == 0`	Varsayılan çıktı	`mr_multiinout_out1`
`count % 3 == 1`	`out1` etiketi	`mr_multiinout_out2`, `a=1/b=1` bölümü
`count % 3 == 2`	`out2` etiketi	`mr_multiinout_out2`, `a=2/b=2` bölümü

cleanup() yöntemi ayrıca tüm anahtarlar işlendikten sonra her çıktıya bir sabit kayıt yazar: ("default", 1L), ("out1", 1L) ve ("out2", 1L).

mr_multiinout_out1 — cleanup() yönteminden ("default", 1L) kaydını alır. Test verilerinden hiçbir anahtar count % 3 == 0 koşulunu karşılamaz, bu nedenle bu tablo yalnızca bir satır içerir:

+------------+------------+
| key        | cnt        |
+------------+------------+
| default    | 1          |
+------------+------------+

mr_multiinout_out2 — a=1/b=1 bölümünde count % 3 == 1 koşuluna uyan anahtarları ve a=2/b=2 bölümünde count % 3 == 2 koşuluna uyan anahtarları alır; ayrıca her bölüm için sabit cleanup() kayıtlarını da içerir:

+--------+------------+---+---+
| key    | cnt        | a | b |
+--------+------------+---+---+
| odps   | 1          | 1 | 1 |
| world  | 1          | 1 | 1 |
| out1   | 1          | 1 | 1 |
| hello  | 2          | 2 | 2 |
| out2   | 1          | 2 | 2 |
+--------+------------+---+---+

Örnek kod

Project Object Model (POM) bağımlılık yapılandırması için Önlemler bölümüne bakın.

Örnek üç ana bölümden oluşur: mapper, çoklu çıkış yönlendirmeli reducer ve giriş/çıkış tablosu bağımsız değişkenlerini ayrıştıran main yöntemi.

Mapper: giriş kayıtlarını belirteçlere ayırın

TokenizerMapper, her alan değerini 1 sayısı ile birlikte bir anahtar olarak yayar.

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

Reducer: Çıktıyı etikete göre yönlendirin

SumReducer, setup() içinde etiketli her hedef için bir çıktı kaydı oluşturur. reduce() içinde, her anahtarı count % 3 değerine göre bir hedefe yönlendirir. cleanup() yöntemi, tüm anahtarlar işlendikten sonra her çıktıya sabit bir kayıt yazar.

Çoklu çıkış yönlendirmesi için temel API'ler şunlardır:

API	Amaç
`context.createOutputRecord()`	Varsayılan (etiketsiz) çıktı için bir kayıt oluşturun
`context.createOutputRecord("out1")`	`"out1"` etiketli çıktı için bir kayıt oluşturun
`context.write(result)`	Varsayılan çıktıya yazın
`context.write(result1, "out1")`	`"out1"` etiketli çıktıya yazın

public static class SumReducer extends ReducerBase {
    private Record result;
    private Record result1;
    private Record result2;

    @Override
    public void setup(TaskContext context) throws IOException {
        /** Create a record for each output and add labels to distinguish outputs. */
        result = context.createOutputRecord();          // default output (no label)
        result1 = context.createOutputRecord("out1");  // output labeled "out1"
        result2 = context.createOutputRecord("out2");  // output labeled "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);
            /** If you do not specify a label, the default output is used. */
            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");
    }
}

Main yöntemi: Girişleri ve çıktıları yapılandırın

main yöntemi, komut satırı bağımsız değişkenlerinden giriş ve çıkış tablosu dizelerini ayrıştırır ve ardından işi gönderir. Giriş girdileri table_name veya table_name|partition_spec biçimini kullanır. Çıkış girdileri isteğe bağlı bir etiket ekler: table_name|partition_spec|label.

/** Convert partition strings such as "ds=1/pt=2" to 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"));

    /** Parse input table strings. */
    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);
        }
    }

    /** Parse output table strings. */
    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);
}

Bu örnek için tam içe aktarmalar:

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;