Ejemplo de uso de recursos - MaxCompute - Centro de documentación de Alibaba Cloud

Funcionamiento

El mapper lee el archivo de texto durante la fase de configuración mediante context.readResourceFileAsStream(). Luego analiza sintácticamente cada línea y escribe los registros en la tabla de salida. El trabajo no tiene fase de reduce.

La bandera -resources y la bandera -classpath tienen propósitos distintos:

Flag	Purpose
`-resources mapreduce-examples.jar,import.txt`	Registra los archivos para que MaxCompute los distribuya a los nodos worker. Úsela con cualquier archivo que el trabajo necesite leer durante la ejecución.
`-classpath data\resources\mapreduce-examples.jar`	Agrega el archivo JAR al classpath de Java para que la JVM cargue la clase `Upload`. Úsela con los archivos JAR que contienen el código del trabajo.

Requisitos previos

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

Haber completado la configuración del entorno descrita en Quick start.
El archivo mapreduce-examples.jar en el directorio bin\data\resources de su instalación local de MaxCompute.

Preparación de tablas y recursos

Ejecute los comandos a continuación en el cliente de MaxCompute.

Cree la tabla de salida:

CREATE TABLE mr_upload_src(key BIGINT, value STRING);

Agregue el archivo de texto y el paquete JAR como recursos:
```
add file data\resources\import.txt -f;
add jar data\resources\mapreduce-examples.jar -f;
```
Si es la primera vez que agrega el paquete JAR, puede omitir la bandera -f.

El archivo import.txt contiene los datos a continuación:
```
1000,odps
```

Ejecución del trabajo

Ejecute el comando a continuación en el cliente de MaxCompute para iniciar el trabajo de carga:

jar -resources mapreduce-examples.jar,import.txt -classpath data\resources\mapreduce-examples.jar
com.aliyun.odps.mapred.open.example.Upload import.txt mr_upload_src;

Referencia de parámetros:

Parameter	Description
`-resources mapreduce-examples.jar,import.txt`	Registra el archivo JAR y el archivo de texto para que MaxCompute los distribuya a los nodos worker.
`-classpath data\resources\mapreduce-examples.jar`	Agrega el archivo JAR al classpath de Java para que se cargue la clase `Upload`.
`import.txt`	Primer argumento de `Upload`: el nombre del archivo de recurso. Se pasa al trabajo mediante `import.filename` en `JobConf`.
`mr_upload_src`	Segundo argumento de `Upload`: la tabla de salida.

Comprobación del resultado

Una vez que finalice el trabajo, consulte mr_upload_src. La tabla debe contener la fila a continuación, que corresponde directamente a la línea 1000,odps del archivo import.txt (campo 0 → key, campo 1 → value):

+------------+------------+
| key        | value      |
+------------+------------+
| 1000       | odps       |
+------------+------------+

Código de ejemplo

Para conocer la configuración de dependencias del modelo de objetos del proyecto (POM), consulte Quick start.

package com.aliyun.odps.mapred.open.example;
import java.io.BufferedInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
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.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;

/**
 * Upload
 * Import data from a text file into a table.
 */
public class Upload {
    public static class UploadMapper extends MapperBase {
        @Override
        public void setup(TaskContext context) throws IOException {
            Record record = context.createOutputRecord();
            StringBuilder importdata = new StringBuilder();
            BufferedInputStream bufferedInput = null;
            try {
                byte[] buffer = new byte[1024];
                int bytesRead = 0;
                // Get the resource file name from JobConf
                String filename = context.getJobConf().get("import.filename");
                // Read the resource file distributed to this worker node
                bufferedInput = context.readResourceFileAsStream(filename);
                while ((bytesRead = bufferedInput.read(buffer)) != -1) {
                    String chunk = new String(buffer, 0, bytesRead);
                    importdata.append(chunk);
                }
                // Parse each line: split by comma, write key (BIGINT) and value (STRING)
                String lines[] = importdata.toString().split("\n");
                for (int i = 0; i < lines.length; i++) {
                    String[] ss = lines[i].split(",");
                    record.set(0, Long.parseLong(ss[0].trim()));
                    record.set(1, ss[1].trim());
                    context.write(record);
                }
            } catch (FileNotFoundException ex) {
                throw new IOException(ex);
            } catch (IOException ex) {
                throw new IOException(ex);
            } finally {
            }
        }

        @Override
        public void map(long recordNum, Record record, TaskContext context)
            throws IOException {
        }
    }

    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: Upload <import_txt> <out_table>");
            System.exit(2);
        }
        JobConf job = new JobConf();
        job.setMapperClass(UploadMapper.class);
        // Pass the resource file name to the mapper via JobConf
        job.set("import.filename", args[0]);
        // Set reducers to 0: this is a map-only job
        job.setNumReduceTasks(0);
        job.setMapOutputKeySchema(SchemaUtils.fromString("key:bigint"));
        job.setMapOutputValueSchema(SchemaUtils.fromString("value:string"));
        InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), job);
        OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
        JobClient.runJob(job);
    }
}

Configuración de JobConf

El ejemplo anterior utiliza la interfaz JobConf del SDK para establecer las propiedades del trabajo. Como alternativa, use el parámetro -conf en el comando jar para especificar un archivo de configuración de JobConf:

jar -conf <your-jobconf-file> ...