En un trabajo map-only, el mapper escribe los registros de salida directamente en una tabla de MaxCompute; no se ejecuta ningún reducer. A diferencia del MapReduce estándar, solo necesita especificar las tablas de salida, sin metadatos de clave-valor para la salida del mapper.
Este ejemplo demuestra tres aspectos:
Cómo configurar un trabajo map-only al establecer el número de reducers en 0
Cómo pasar parámetros mediante JobConf y leerlos dentro del mapper
Cómo funcionan los métodos del ciclo de vida
setup,mapycleanupcon ejecución condicional
Requisitos previos
Antes de empezar, complete la configuración del entorno descrita en Primeros pasos.
Preparar las tablas y los recursos de prueba
-
Cree las tablas de entrada y de salida.
CREATE TABLE wc_in (key STRING, value STRING); CREATE TABLE wc_out (key STRING, cnt BIGINT); -
Agregue el paquete JAR como recurso.
add jar data\resources\mapreduce-examples.jar -f;Omita
-fla primera vez que agregue el paquete JAR. La rutadata\resources\mapreduce-examples.jares relativa albindirectorio de la instalación local de su cliente de MaxCompute. -
Importe los datos de prueba a
wc_inmediante Tunnel. Ejecute el siguiente comando desde el directoriobindel cliente de MaxCompute, donde se encuentradata.txt.tunnel upload data.txt wc_in;El comando carga las siguientes filas en
wc_in:hello,odps hello,odps
Ejecutar el trabajo
Ejecute el siguiente comando en el cliente de MaxCompute:
jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.MapOnly wc_in wc_out map
Los argumentos del comando se asignan de la siguiente manera:
|
Argument |
Description |
|
|
Declara el paquete JAR como dependencia del trabajo |
|
|
Especifica la ruta al paquete JAR |
|
|
Tabla de entrada |
|
|
Tabla de salida |
|
|
Establece |
Resultado esperado
Una vez finalizado el trabajo, consulte wc_out:
+------------+------------+
| key | cnt |
+------------+------------+
| hello | 1 |
| hello | 1 |
+------------+------------+
La tabla contiene dos filas en lugar de una porque un trabajo map-only produce un registro de salida por cada registro de entrada, sin agregación. Cada fila de entrada hello,odps se asigna a una fila de salida hello | 1.
Código de ejemplo
Para conocer las dependencias del Project Object Model (POM), consulte Precauciones.
package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.conf.JobConf;
import com.aliyun.odps.mapred.utils.SchemaUtils;
import com.aliyun.odps.mapred.utils.InputUtils;
import com.aliyun.odps.mapred.utils.OutputUtils;
import com.aliyun.odps.data.TableInfo;
public class MapOnly {
public static class MapperClass extends MapperBase {
@Override
public void setup(TaskContext context) throws IOException {
boolean is = context.getJobConf().getBoolean("option.mapper.setup", false);
/** The main function executes the following logic only if option.mapper.setup is set to true in the JobConf file: */
if (is) {
Record result = context.createOutputRecord();
result.set(0, "setup");
result.set(1, 1L);
context.write(result);
}
}
@Override
public void map(long key, Record record, TaskContext context) throws IOException {
boolean is = context.getJobConf().getBoolean("option.mapper.map", false);
/** The main function executes the following logic only if option.mapper.map is set to true in the JobConf file: */
if (is) {
Record result = context.createOutputRecord();
result.set(0, record.get(0));
result.set(1, 1L);
context.write(result);
}
}
@Override
public void cleanup(TaskContext context) throws IOException {
boolean is = context.getJobConf().getBoolean("option.mapper.cleanup", false);
/** The main function executes the following logic only if option.mapper.cleanup is set to true in the JobConf file: */
if (is) {
Record result = context.createOutputRecord();
result.set(0, "cleanup");
result.set(1, 1L);
context.write(result);
}
}
}
public static void main(String[] args) throws Exception {
if (args.length != 2 && args.length != 3) {
System.err.println("Usage: OnlyMapper <in_table> <out_table> [setup|map|cleanup]");
System.exit(2);
}
JobConf job = new JobConf();
job.setMapperClass(MapperClass.class);
/** For MapOnly jobs, the number of reducers must be explicitly set to 0. */
job.setNumReduceTasks(0);
/** Configure information about input and output tables. */
InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
if (args.length == 3) {
String options = new String(args[2]);
/** You can specify key-value pairs in the JobConf file, and use getJobConf of the context to query the configurations in a mapper. */
if (options.contains("setup")) {
job.setBoolean("option.mapper.setup", true);
}
if (options.contains("map")) {
job.setBoolean("option.mapper.map", true);
}
if (options.contains("cleanup")) {
job.setBoolean("option.mapper.cleanup", true);
}
}
JobClient.runJob(job);
}
}
Cómo funciona el código
main(): configuración del trabajo
job.setNumReduceTasks(0) es obligatorio para los trabajos map-only. Sin esta instrucción, el framework espera un reducer y el trabajo falla. InputUtils.addTable y OutputUtils.addTable vinculan las tablas de entrada y de salida al trabajo mediante los argumentos de la línea de comandos.
El tercer argumento opcional (setup, map o cleanup) establece la bandera booleana correspondiente en JobConf. Cada método del ciclo de vida lee su propia bandera mediante context.getJobConf().getBoolean(...) y escribe la salida únicamente cuando la bandera es true.
setup(): se ejecuta una vez antes de iniciar el procesamiento
Escribe un único registro con la clave "setup" y el conteo 1. Se ejecuta cuando option.mapper.setup=true.
map(): se ejecuta una vez por cada registro de entrada
Lee el primer campo de cada registro de entrada (record.get(0)) y lo escribe con el conteo 1. Se ejecuta cuando option.mapper.map=true. En este ejemplo, el argumento map habilita este método, razón por la cual la salida contiene dos filas: una por cada fila de entrada.
cleanup(): se ejecuta una vez después de procesar todos los registros
Escribe un único registro con la clave "cleanup" y el conteo 1. Se ejecuta cuando option.mapper.cleanup=true.