Ejemplo de MultiJobs - MaxCompute - Centro de documentación de Alibaba Cloud

Este ejemplo muestra cómo encadenar varios trabajos de MapReduce de forma secuencial en MaxCompute. Un solo trabajo de MapReduce no puede invocar a otro durante la ejecución. Por lo tanto, cuando su lógica de procesamiento requiere una ejecución iterativa, donde cada iteración depende del resultado de la anterior, debe orquestar varios trabajos desde un método main del lado del cliente. Este ejemplo utiliza un contador para controlar el bucle de iteración.

Cómo funciona

El programa ejecuta dos tipos de trabajos en secuencia:

Trabajo de inicialización (InitMapper): escribe un valor inicial de contador (2) en la tabla de salida.
Trabajos de decremento (DecreaseMapper): se ejecutan dentro de un bucle. Cada iteración lee el resultado anterior desde una tabla de recursos, resta 1 al valor y establece un contador. El bucle termina cuando el contador llega a 0.

Para pasar datos entre trabajos se usan dos mecanismos:

Tabla de recursos (multijobs_res_table): transfiere el valor numérico entre los trabajos.
Contador (multijobs.value): indica al método main si debe continuar con las iteraciones.

El método main es el único encargado de la orquestación: envía los trabajos en orden y comprueba que cada uno haya terminado antes de avanzar al siguiente paso.

Requisitos previos

Antes de empezar, asegúrese de haber cumplido con lo siguiente:

Completó la configuración del entorno descrita en Primeros pasos
Compiló el paquete mapreduce-examples.jar y lo colocó en bin\data\resources dentro de la ruta de instalación del cliente de MaxCompute

Preparar tablas y recursos

Cree las tablas de prueba:

CREATE TABLE mr_empty (key STRING, value STRING);
CREATE TABLE mr_multijobs_out (value BIGINT);

Registre los recursos que utilizará el trabajo:
```
add table mr_multijobs_out as multijobs_res_table -f;

-- Omit -f when adding the JAR for the first time.
add jar data\resources\mapreduce-examples.jar -f;
```
La tabla mr_multijobs_out se registra como multijobs_res_table para que DecreaseMapper pueda leer el resultado del trabajo anterior como una tabla de recursos. El archivo JAR se registra para que el cliente de MaxCompute localice la clase durante la ejecución.

Ejecutar MultiJobs

Ejecute el siguiente comando en el cliente de MaxCompute:

jar -resources mapreduce-examples.jar,multijobs_res_table -classpath data\resources\mapreduce-examples.jar \
    com.aliyun.odps.mapred.open.example.MultiJobs mr_multijobs_out;

Parámetros del comando:

Parameter	Value	Description
`-resources`	`mapreduce-examples.jar,multijobs_res_table`	Recursos disponibles para el trabajo: el paquete JAR y la tabla de recursos que contiene el resultado del trabajo anterior
`-classpath`	`data\resources\mapreduce-examples.jar`	Ruta local del archivo JAR que usa el cliente para ubicar la clase principal
Clase principal	`com.aliyun.odps.mapred.open.example.MultiJobs`	Punto de entrada del programa
Argumento	`mr_multijobs_out`	Nombre de la tabla de salida que se pasa a `main`

Resultado esperado

Una vez que el trabajo termina, mr_multijobs_out contiene un registro:

+------------+
| value      |
+------------+
| 0          |
+------------+

Código de ejemplo

Para conocer la configuración de dependencias del Project Object Model (POM), consulte la sección Precauciones de la guía de primeros pasos.

package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import java.util.Iterator;
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.RunningJob;
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;
/**
     * MultiJobs
     *
     * Running multiple job
     *
     **/
public class MultiJobs {
    public static class InitMapper extends MapperBase {
        @Override
            public void setup(TaskContext context) throws IOException {
            Record record = context.createOutputRecord();
            long v = context.getJobConf().getLong("multijobs.value", 2);
            record.set(0, v);
            context.write(record);
        }
    }
    public static class DecreaseMapper extends MapperBase {
        @Override
            public void cleanup(TaskContext context) throws IOException {
            /** Obtain the variable values that are defined in the main function from JobConf. */
            long expect = context.getJobConf().getLong("multijobs.expect.value", -1);
            long v = -1;
            int count = 0;
            /** Read the data from the output table of the previous job. */
            Iterator<Record> iter = context.readResourceTable("multijobs_res_table");
            while (iter.hasNext()) {
                Record r = iter.next();
                v = (Long) r.get(0);
                if (expect != v) {
                    throw new IOException("expect: " + expect + ", but: " + v);
                }
                count++;
            }
            if (count != 1) {
                throw new IOException("res_table should have 1 record, but: " + count);
            }
            Record record = context.createOutputRecord();
            v--;
            record.set(0, v);
            context.write(record);
            /** Set the counter. The counter value can be obtained in the main function after the job is completed. */
            context.getCounter("multijobs", "value").setValue(v);
        }
    }
    public static void main(String[] args) throws Exception {
        if (args.length != 1) {
            System.err.println("Usage: TestMultiJobs <table>");
            System.exit(1);
        }
        String tbl = args[0];
        long iterCount = 2;
        System.err.println("Start to run init job.");
        JobConf initJob = new JobConf();
        initJob.setLong("multijobs.value", iterCount);
        initJob.setMapperClass(InitMapper.class);
        InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), initJob);
        OutputUtils.addTable(TableInfo.builder().tableName(tbl).build(), initJob);
        initJob.setMapOutputKeySchema(SchemaUtils.fromString("key:string"));
        initJob.setMapOutputValueSchema(SchemaUtils.fromString("value:string"));
        /** Explicitly set the number of reducers to 0 for map-only jobs. */
        initJob.setNumReduceTasks(0);
        JobClient.runJob(initJob);
        while (true) {
            System.err.println("Start to run iter job, count: " + iterCount);
            JobConf decJob = new JobConf();
            decJob.setLong("multijobs.expect.value", iterCount);
            decJob.setMapperClass(DecreaseMapper.class);
            InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), decJob);
            OutputUtils.addTable(TableInfo.builder().tableName(tbl).build(), decJob);
            /** Explicitly set the number of reducers to 0 for map-only jobs. */
            decJob.setNumReduceTasks(0);
            RunningJob rJob = JobClient.runJob(decJob);
            iterCount--;
            /** If the specified number of iterations is reached, exit the loop. */
            if (rJob.getCounters().findCounter("multijobs", "value").getValue() == 0) {
                break;
            }
        }
        if (iterCount != 0) {
            throw new IOException("Job failed.");
        }
    }
}

Explicación del código

InitMapper

Se ejecuta una sola vez en setup(). Lee el valor inicial del contador (multijobs.value, cuyo valor predeterminado es 2) desde JobConf y lo escribe en la tabla de salida. Se trata de un trabajo exclusivo de map, por lo que setNumReduceTasks(0) evita que el framework inicie algún reducer.

DecreaseMapper

Se ejecuta en cleanup() al final de cada iteración. Realiza las siguientes acciones:

Lee el valor esperado (multijobs.expect.value) desde JobConf.
Lee el único registro de multijobs_res_table mediante context.readResourceTable(). Lanza una excepción IOException si la cantidad de registros no es exactamente 1 o si el valor no coincide con el esperado.
Resta 1 al valor, lo escribe en la tabla de salida y actualiza el contador (multijobs.value) con el nuevo valor.

main método

Controla la secuencia de los trabajos:

Envía el trabajo de inicialización y espera a que termine.
Inicia un bucle que envía los trabajos de decremento uno por uno. Espera a que cada uno finalice antes de continuar.
Después de cada trabajo de decremento, comprueba el contador multijobs.value. Si es igual a 0, sale del bucle.
Verifica que iterCount sea 0 al terminar el bucle. De lo contrario, lanza una excepción IOException para indicar que el trabajo terminó con errores.