Examples of using partitioned tables as input - MaxCompute - Alibaba Cloud Dokümantasyon Merkezi

Bu örnekler, bir MaxCompute MapReduce işinde belirli bölümlerden veri okumak için bölüm belirtimiyle birlikte InputUtils.addTable() öğesinin nasıl kullanılacağını gösterir.

Her iki örnekte de yalnızca main işlevi gösterilir. Kod eksiktir ve doğrudan derlenip çalıştırılamaz; kendi uygulamanızı oluştururken referans olarak kullanın.

Örnek 1: Tek bir bölümden okuma

Bölüm değeri iş gönderimi sırasında biliniyorsa bu kalıbı kullanın.

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

Örnek 2: Birden çok bölümden dinamik olarak okuma

Çalışma zamanında bölümleri filtrelemeniz gerekiyorsa bu kalıbı kullanın. Bu örnekte MaxCompute SDK ve MapReduce SDK birlikte kullanılır: MaxCompute SDK tablodaki tüm bölümleri listeler ve özel bir applicable işlevi hangi bölümlerin giriş olarak dahil edileceğini belirler.

applicable işlevi, gereksinimlerinize göre bölümleri filtrelemek için uyguladığınız özel mantıktır.

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

Bölüm döngüsünün çalışma şekli:

odps.tables().get(tblname) çağrısı, MaxCompute SDK'yı kullanarak tüm bölümler dahil olmak üzere tablo meta verilerini alır.
for döngüsü her bir bölüm üzerinde yinelenir ve bölümün dahil edilip edilmeyeceğine karar vermek için sizin uyguladığınız özel bir işlev olan applicable(p) işlevini çağırır.
Dahil edilen her bölüm için, bölümün anahtar-değer çiftlerinden bir LinkedHashMap<String, String> oluşturulur ve bölüm belirtimi olarak InputUtils.addTable() yöntemine iletilir.
Çıkış tablosu, OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job) kullanılarak bir bölüm belirtimi olmadan eklenir.