Este exemplo demonstra como executar um job MapReduce MultipleInOut no MaxCompute. O job lê dados de duas tabelas de entrada e direciona a saída para duas tabelas distintas — uma não particionada e outra particionada — com rótulos de saída que controlam o destino de escrita de cada reducer.
Pré-requisitos
Antes de começar, verifique se você tem:
Concluído a configuração do ambiente descrita em Primeiros passos
O arquivo
mapreduce-examples.jar, armazenado no diretóriobin\data\resourcesda instalação do seu cliente MaxCompute
Configure as tabelas de teste e os recursos
Crie as tabelas
Execute as instruções a seguir no cliente MaxCompute para criar duas tabelas de entrada e duas de saída. A tabela mr_multiinout_out2 é particionada; adicione duas partições antes de executar o job.
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');Adicione o 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;Faça upload dos dados de teste
Use o Tunnel para fazer upload dos arquivos data1.txt e data2.txt do diretório bin do cliente MaxCompute para as duas tabelas de entrada.
tunnel upload data1.txt wc_in1;
tunnel upload data2.txt wc_in2;Após o upload, as tabelas contêm os seguintes dados:
wc_in1:
hello,odpswc_in2:
hello,worldExecute o job
Execute o comando a seguir no cliente 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;O comando aceita dois argumentos posicionais:
|
Argumento |
Formato |
Exemplo |
||||
|
Entradas |
Nomes de tabelas separados por vírgula |
|
||||
|
Saídas |
Entradas separadas por vírgula, cada uma no formato |
partition_spec |
label` |
|
a=1/b=1 |
out1` |
Nas saídas, partition_spec e label são opcionais. Se você omitir label, a saída será direcionada ao destino padrão (sem rótulo).
Resultados esperados
O reducer encaminha cada chave para uma tabela de saída específica conforme o resto da divisão da contagem total da chave por 3:
|
Condição |
Destino |
Tabela de saída |
|
|
Saída padrão |
|
|
|
Rótulo |
|
|
|
Rótulo |
|
O método cleanup() também grava um registro fixo em cada saída após o processamento de todas as chaves: ("default", 1L), ("out1", 1L) e ("out2", 1L).
mr_multiinout_out1 — recebe o registro ("default", 1L) de cleanup(). Nenhuma chave dos dados de teste satisfaz a condição count % 3 == 0, portanto esta tabela contém apenas uma linha:
+------------+------------+
| key | cnt |
+------------+------------+
| default | 1 |
+------------+------------+mr_multiinout_out2 — recebe as chaves com count % 3 == 1 na partição a=1/b=1 e as chaves com count % 3 == 2 na partição a=2/b=2, além dos registros fixos de cleanup() para cada partição:
+--------+------------+---+---+
| 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 exemplo
Para configurar as dependências do Project Object Model (POM), consulte Precauções.
O exemplo divide-se em três partes principais: o mapper, o reducer com roteamento de múltiplas saídas e o método main, que analisa os argumentos das tabelas de entrada e saída.
Mapper: tokenização dos registros de entrada
A classe TokenizerMapper emite cada valor de campo como uma chave com contagem 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: roteamento de saída por rótulo
A classe SumReducer cria um registro de saída para cada destino rotulado durante o método setup(). No método reduce(), ela encaminha cada chave para um destino específico com base no resultado de count % 3. O método cleanup() grava um registro fixo em cada saída após a conclusão do processamento de todas as chaves.
As principais APIs para o roteamento de múltiplas saídas são:
|
API |
Finalidade |
|
|
Cria um registro para a saída padrão (sem rótulo) |
|
|
Cria um registro para a saída rotulada como |
|
|
Grava na saída padrão |
|
|
Grava na saída rotulada como |
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: configuração de entradas e saídas
O método main analisa as strings das tabelas de entrada e saída nos argumentos da linha de comando e envia o job. As entradas usam o formato table_name ou table_name|partition_spec. As saídas incluem um rótulo 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);
}Imports completos deste exemplo:
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;