MultipleInOut example - MaxCompute - Centro de documentación de Alibaba Cloud

Requisitos previos

Antes de empezar, asegúrese de cumplir con lo siguiente:

Haber completado la configuración del entorno descrita en Primeros pasos
Contar con el archivo mapreduce-examples.jar, almacenado en el directorio bin\data\resources de su instalación del cliente de MaxCompute

Configurar tablas y recursos de prueba

Crear las tablas

Ejecute las siguientes sentencias en su cliente de MaxCompute para crear dos tablas de entrada y dos de salida. La tabla mr_multiinout_out2 está particionada; agregue dos particiones antes de ejecutar el trabajo.

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

Agregar el recurso JAR

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

Cargar datos de prueba

Utilice Tunnel para cargar los archivos data1.txt y data2.txt desde el directorio bin de su cliente de MaxCompute hacia las dos tablas de entrada.

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

Después de la carga, las tablas contienen los siguientes datos:

wc_in1:

hello,odps

wc_in2:

hello,world

Ejecutar el trabajo

Ejecute el siguiente comando en su cliente de MaxCompute:

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;

El comando acepta dos argumentos posicionales:

Argument	Format	Example
Entradas	Nombres de tablas separados por comas	`wc_in1,wc_in2`
Salidas	Entradas separadas por comas, cada una con el formato `table_name\|partition_spec\|label`	`mr_multiinout_out2\|a=1/b=1\|out1`

Para las salidas, los parámetros partition_spec y label son opcionales. Si omite label, la salida se dirigirá al destino predeterminado (sin etiqueta).

Resultados esperados

El reducer asigna cada clave a una tabla de salida específica según el conteo total de la clave módulo 3:

Condition	Destination	Output table
`count % 3 == 0`	Salida predeterminada	`mr_multiinout_out1`
`count % 3 == 1`	Etiqueta `out1`	`mr_multiinout_out2`, partición `a=1/b=1`
`count % 3 == 2`	Etiqueta `out2`	`mr_multiinout_out2`, partición `a=2/b=2`

El método cleanup() también escribe un registro fijo en cada salida después de procesar todas las claves: ("default", 1L), ("out1", 1L) y ("out2", 1L).

mr_multiinout_out1: recibe el registro ("default", 1L) proveniente de cleanup(). Dado que ninguna clave de los datos de prueba cumple con count % 3 == 0, esta tabla contiene solo una fila:

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

mr_multiinout_out2: recibe las claves con count % 3 == 1 en la partición a=1/b=1 y las claves con count % 3 == 2 en la partición a=2/b=2, además de los registros fijos de cleanup() para cada partición:

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

Código de ejemplo

Para consultar la configuración de dependencias del Project Object Model (POM), consulte Precauciones.

El ejemplo consta de tres partes principales: el mapper, el reducer con enrutamiento de salida múltiple y el método main, que analiza los argumentos de las tablas de entrada y salida.

Mapper: tokenizar registros de entrada

TokenizerMapper emite cada valor de campo como una clave con un conteo de 1.

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: enrutar salida por etiqueta

SumReducer crea un registro de salida por cada destino etiquetado dentro de setup(). En reduce(), asigna cada clave a un destino basándose en count % 3. El método cleanup() escribe un registro fijo en cada salida una vez procesadas todas las claves.

Las API clave para el enrutamiento de salida múltiple son:

API	Purpose
`context.createOutputRecord()`	Crea un registro para la salida predeterminada (sin etiqueta)
`context.createOutputRecord("out1")`	Crea un registro para la salida etiquetada como `"out1"`
`context.write(result)`	Escribe en la salida predeterminada
`context.write(result1, "out1")`	Escribe en la salida etiquetada como `"out1"`

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

Método main: configurar entradas y salidas

El método main analiza las cadenas de las tablas de entrada y salida a partir de los argumentos de la línea de comandos y luego envía el trabajo. Las entradas utilizan el formato table_name o table_name|partition_spec. Las salidas incluyen una etiqueta opcional: 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);
}

Importaciones completas para este ejemplo:

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;