Examples of using partitioned tables as input - MaxCompute: Central de Documentação do Alibaba Cloud

Estes exemplos mostram como usar InputUtils.addTable() com uma especificação de partição para ler dados de partições específicas em um job do MaxCompute MapReduce.

Ambos os exemplos mostram apenas a função main. O código está incompleto e não é possível compilá-lo nem executá-lo diretamente — use-o como referência ao desenvolver sua própria implementação.

Exemplo 1: Ler de uma única partição

Use este padrão quando o valor da partição for conhecido no momento do envio do job.

public static void main(String[] args) throws Exception {
    JobConf job = new JobConf();
    ...
    LinkedHashMap<String, String> input = new LinkedHashMap<String, String>();
    input.put("pt", "123456");
    InputUtils.addTable(TableInfo.builder().tableName("input_table").partSpec(input).build(), job);
    LinkedHashMap<String, String> output = new LinkedHashMap<String, String>();
    output.put("ds", "654321");
    OutputUtils.addTable(TableInfo.builder().tableName("output_table").partSpec(output).build(), job);
    JobClient.runJob(job);
}

Exemplo 2: Ler de várias partições dinamicamente

Use este padrão quando precisar filtrar partições em tempo de execução. Este exemplo combina o SDK do MaxCompute e o SDK do MapReduce: o SDK do MaxCompute lista todas as partições da tabela, e uma função applicable personalizada determina quais partições incluir como entrada.

A função applicable é uma lógica personalizada que você implementa para filtrar partições de acordo com seus requisitos.

package com.aliyun.odps.mapred.open.example;
...
    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);
    job.setMapOutputKeySchema(SchemaUtils.fromString("word:string"));
    job.setMapOutputValueSchema(SchemaUtils.fromString("count:bigint"));
    // Using an Alibaba Cloud account's AccessKey pair grants access to all API operations,
    // which is a high-risk approach. Use a RAM user instead for routine operations.
    // To create a RAM user, go to the Resource Access Management (RAM) console.
    // Store credentials in environment variables rather than hardcoding them in your code.
    Account account = new AliyunAccount(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"), System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
    Odps odps = new Odps(account);
    odps.setEndpoint("odps_endpoint_url");
    odps.setDefaultProject("my_project");
    Table table = odps.tables().get(tblname);
    TableInfoBuilder builder = TableInfo.builder().tableName(tblname);
    for (Partition p : table.getPartitions()) {
        if (applicable(p)) {
            LinkedHashMap<String, String> partSpec = new LinkedHashMap<String, String>();
            for (String key : p.getPartitionSpec().keys()) {
                partSpec.put(key, p.getPartitionSpec().get(key));
            }
            InputUtils.addTable(builder.partSpec(partSpec).build(), job);
        }
    }
    OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
    JobClient.runJob(job);
}

Como funciona o loop de partições:

odps.tables().get(tblname) usa o SDK do MaxCompute para recuperar os metadados da tabela, incluindo todas as partições.
O loop for itera sobre cada partição e chama applicable(p) — uma função personalizada que você implementa — para decidir se deve incluí-la.
Para cada partição incluída, um LinkedHashMap<String, String> é construído a partir dos pares de chave-valor da partição e passado para InputUtils.addTable() como a especificação de partição.
A tabela de saída é adicionada sem especificação de partição usando OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job).