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

Este ejemplo muestra cómo ejecutar un trabajo de WordCount en MaxCompute MapReduce. Abarca desde la creación de las tablas de entrada y la carga de datos hasta la ejecución del trabajo y la comprobación de los resultados.

Cómo funciona

El ejemplo de WordCount sigue el flujo de datos estándar de MapReduce:

(input) <k1, v1>  ->  map  ->  <k2, v2>  ->  combine  ->  <k2, v2>  ->  reduce  ->  <k3, v3>  (output)

Tres clases implementan este flujo:

TokenizerMapper: lee cada registro de la tabla de entrada y emite el valor de cada campo como una clave con un conteo de 1.
SumCombiner: agrega los conteos de forma local en el nodo del mapper antes de enviar los datos al reducer, lo que reduce el tráfico de red.
SumReducer: suma todos los conteos para cada clave única y escribe el resultado final en la tabla de salida.

Requisitos previos

Antes de empezar, asegúrese de haber completado lo siguiente:

La configuración del entorno descrita en Primeros pasos.

Preparar el entorno

1. Prepare el paquete JAR.

El paquete JAR de este ejemplo es mapreduce-examples.jar, ubicado en el directorio bin\data\resources de la ruta de instalación del cliente de MaxCompute.

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 -f cuando agregue el paquete JAR por primera vez.

2. Cargue los datos de entrada.

Use Tunnel para cargar data.txt desde el directorio bin del cliente de MaxCompute hacia la tabla wc_in:

tunnel upload data.txt wc_in;

Los siguientes datos se cargan en wc_in:

hello,odps

Ejecutar WordCount

Ejecute el trabajo de WordCount en el cliente de MaxCompute:

jar -resources mapreduce-examples.jar -classpath data\resources\mapreduce-examples.jar com.aliyun.odps.mapred.open.example.WordCount wc_in wc_out

Comprobar el resultado

Una vez que finalice el trabajo, consulte la tabla wc_out para ver los resultados:

SELECT * FROM wc_out;

El resultado debe coincidir con lo siguiente:

+------------+------------+
| key        | cnt        |
+------------+------------+
| hello      | 1          |
| odps       | 1          |
+------------+------------+

Código de ejemplo

Para configurar las dependencias del Project Object Model (POM), consulte la sección Precautions en 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.ReducerBase;
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;

public class WordCount {

    // Mapper: reads each record, emits (word, 1) for every field value
    public static class TokenizerMapper extends MapperBase {
        private Record word;
        private Record one;

        @Override
        public void setup(TaskContext context) throws IOException {
            word = context.createMapOutputKeyRecord();
            one = context.createMapOutputValueRecord();
            one.set(new Object[] { 1L });
            System.out.println("TaskID:" + context.getTaskID().toString());
        }

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

    // Combiner: aggregates map output locally before shuffling to the reducer
    public static class SumCombiner extends ReducerBase {
        private Record count;

        @Override
        public void setup(TaskContext context) throws IOException {
            count = context.createMapOutputValueRecord();
        }

        @Override
        public void reduce(Record key, Iterator<Record> values, TaskContext context)
                throws IOException {
            long c = 0;
            while (values.hasNext()) {
                Record val = values.next();
                c += (Long) val.get(0);
            }
            count.set(0, c);
            context.write(key, count);
        }
    }

    // Reducer: sums counts for each unique key and writes the final output
    public static class SumReducer extends ReducerBase {
        private Record result = null;

        @Override
        public void setup(TaskContext context) throws IOException {
            result = context.createOutputRecord();
        }

        @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);
            }
            result.set(0, key.get(0));
            result.set(1, count);
            context.write(result);
        }
    }

    public static void main(String[] args) throws Exception {
        if (args.length != 2) {
            System.err.println("Usage: WordCount <in_table> <out_table>");
            System.exit(2);
        }

        JobConf job = new JobConf();
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(SumCombiner.class);
        job.setReducerClass(SumReducer.class);

        // Define the intermediate key-value schema between mapper and reducer
        job.setMapOutputKeySchema(SchemaUtils.fromString("word:string"));
        job.setMapOutputValueSchema(SchemaUtils.fromString("count:bigint"));

        // Set input and output tables
        InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
        OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);

        JobClient.runJob(job);
    }
}

Descripción detallada

Para ilustrar el flujo de datos a través del proceso, a continuación se detalla lo que produce cada etapa con la entrada hello,odps:

Salida del mapper: TokenizerMapper emite un par clave-valor por cada campo:

<hello, 1>
<odps,  1>

Salida del combiner: SumCombiner agrega los datos de forma local (en este pequeño ejemplo, los conteos permanecen en 1):

<hello, 1>
<odps,  1>

Salida del reducer: SumReducer escribe los conteos finales sumados en wc_out:

<hello, 1>
<odps,  1>