MapOnly example - MaxCompute - Alibaba Cloud Dokümantasyon Merkezi

Yalnızca haritalama (map-only) işlerinde, mapper çıktı kayıtlarını doğrudan bir MaxCompute tablosuna yazar; reducer çalışmaz. Standart MapReduce'tan farklı olarak, mapper çıktısı için anahtar-değer meta verilerini belirtmenize gerek kalmaz, yalnızca çıktı tablolarını belirtirsiniz.

Bu örnek şu üç konuyu gösterir:

Reducer sayısını 0 olarak ayarlayarak map-only bir işin nasıl yapılandırılacağı
Parametrelerin JobConf üzerinden nasıl aktarılacağı ve mapper içinde nasıl okunacağı
setup, map ve cleanup yaşam döngüsü yöntemlerinin koşullu yürütme ile nasıl çalıştığı

Önkoşullar

Başlamadan önce, Başlangıç bölümünde açıklanan ortam kurulumunu tamamlayın.

Test tablolarını ve kaynaklarını hazırlama

Giriş ve çıktı tablolarını oluşturun.

CREATE TABLE wc_in (key STRING, value STRING);
CREATE TABLE wc_out (key STRING, cnt BIGINT);

JAR paketini kaynak olarak ekleyin.
```
add jar data\resources\mapreduce-examples.jar -f;
```
JAR paketini ilk kez eklerken -f seçeneğini atlayın. data\resources\mapreduce-examples.jar yolu, yerel MaxCompute istemci kurulumunuzun bin dizinine görelidir.
Tunnel kullanarak test verilerini wc_in tablosuna aktarın. Aşağıdaki komutu, data.txt dosyasının bulunduğu MaxCompute istemcisinin bin dizininden çalıştırın.
```
tunnel upload data.txt wc_in;
```
Komut, aşağıdaki satırları wc_in tablosuna yükler:
```
hello,odps
hello,odps
```

İşi çalıştırma

MaxCompute istemcisinde aşağıdaki komutu çalıştırın:

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

Komut bağımsız değişkenleri aşağıdaki şekilde eşleşir:

Bağımsız Değişken	Açıklama
`-resources mapreduce-examples.jar`	JAR paketini iş bağımlılığı olarak bildirir
`-classpath data\resources\mapreduce-examples.jar`	JAR paketinin yolunu belirtir
`wc_in`	Giriş tablosu
`wc_out`	Çıktı tablosu
`map`	JobConf içinde `option.mapper.map=true` ayarını yaparak `map()` yönteminin çıktı yazmasını etkinleştirir

Beklenen sonuç

İş tamamlandıktan sonra wc_out tablosunu sorgulayın:

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

Map-only bir iş toplama yapmadığından ve her giriş kaydı için bir çıktı kaydı ürettiğinden tablo bir yerine iki satır içerir. Her hello,odps giriş satırı, bir hello | 1 çıktı satırına eşlenir.

Örnek kod

Project Object Model (POM) bağımlılıkları için Önlemler bölümüne bakın.

package com.aliyun.odps.mapred.open.example;
import java.io.IOException;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.mapred.JobClient;
import com.aliyun.odps.mapred.MapperBase;
import com.aliyun.odps.mapred.conf.JobConf;
import com.aliyun.odps.mapred.utils.SchemaUtils;
import com.aliyun.odps.mapred.utils.InputUtils;
import com.aliyun.odps.mapred.utils.OutputUtils;
import com.aliyun.odps.data.TableInfo;
public class MapOnly {
    public static class MapperClass extends MapperBase {
        @Override
            public void setup(TaskContext context) throws IOException {
            boolean is = context.getJobConf().getBoolean("option.mapper.setup", false);
            /** The main function executes the following logic only if option.mapper.setup is set to true in the JobConf file: */
            if (is) {
                Record result = context.createOutputRecord();
                result.set(0, "setup");
                result.set(1, 1L);
                context.write(result);
            }
        }
        @Override
            public void map(long key, Record record, TaskContext context) throws IOException {
            boolean is = context.getJobConf().getBoolean("option.mapper.map", false);
            /** The main function executes the following logic only if option.mapper.map is set to true in the JobConf file: */
            if (is) {
                Record result = context.createOutputRecord();
                result.set(0, record.get(0));
                result.set(1, 1L);
                context.write(result);
            }
        }
        @Override
            public void cleanup(TaskContext context) throws IOException {
            boolean is = context.getJobConf().getBoolean("option.mapper.cleanup", false);
            /** The main function executes the following logic only if option.mapper.cleanup is set to true in the JobConf file: */
            if (is) {
                Record result = context.createOutputRecord();
                result.set(0, "cleanup");
                result.set(1, 1L);
                context.write(result);
            }
        }
    }
    public static void main(String[] args) throws Exception {
        if (args.length != 2 && args.length != 3) {
            System.err.println("Usage: OnlyMapper <in_table> <out_table> [setup|map|cleanup]");
            System.exit(2);
        }
        JobConf job = new JobConf();
        job.setMapperClass(MapperClass.class);
        /** For MapOnly jobs, the number of reducers must be explicitly set to 0. */
        job.setNumReduceTasks(0);
        /** Configure information about input and output tables. */
        InputUtils.addTable(TableInfo.builder().tableName(args[0]).build(), job);
        OutputUtils.addTable(TableInfo.builder().tableName(args[1]).build(), job);
        if (args.length == 3) {
            String options = new String(args[2]);
            /** You can specify key-value pairs in the JobConf file, and use getJobConf of the context to query the configurations in a mapper. */
            if (options.contains("setup")) {
                job.setBoolean("option.mapper.setup", true);
            }
            if (options.contains("map")) {
                job.setBoolean("option.mapper.map", true);
            }
            if (options.contains("cleanup")) {
                job.setBoolean("option.mapper.cleanup", true);
            }
        }
        JobClient.runJob(job);
    }
}

Kodun çalışma şekli

main() — iş yapılandırması

Map-only işleri için job.setNumReduceTasks(0) gereklidir. Bu ayar yapılmazsa çerçeve bir reducer bekler ve iş başarısız olur. InputUtils.addTable ve OutputUtils.addTable yöntemleri, giriş ve çıktı tablolarını komut satırı bağımsız değişkenlerini kullanarak işe bağlar.

Üçüncü isteğe bağlı bağımsız değişken (setup, map veya cleanup), JobConf içinde ilgili boolean bayrağını ayarlar. Her yaşam döngüsü yöntemi, kendi bayrağını context.getJobConf().getBoolean(...) aracılığıyla okur ve yalnızca bayrak true olduğunda çıktı yazar.

setup() — işlem başlamadan önce bir kez çalışır

Anahtar değeri "setup" ve sayı değeri 1 olan tek bir kayıt yazar. option.mapper.setup=true olduğunda çalışır.

map() — her giriş kaydı için bir kez çalışır

Her giriş kaydının ilk alanını (record.get(0)) okur ve bunu sayı değeri 1 ile yazar. option.mapper.map=true olduğunda çalışır. Bu örnekte, map bağımsız değişkeni bu yöntemi etkinleştirir; bu nedenle çıktı, her giriş satırı için bir tane olmak üzere iki satır içerir.

cleanup() — tüm kayıtlar işlendikten sonra bir kez çalışır

Anahtar değeri "cleanup" ve sayı değeri 1 olan tek bir kayıt yazar. option.mapper.cleanup=true olduğunda çalışır.