MultiJobs örneği - MaxCompute - Alibaba Cloud Dokümantasyon Merkezi

Bu örnek, MaxCompute üzerinde birden fazla MapReduce işinin nasıl sıralı şekilde zincirleneceğini gösterir. Tek bir MapReduce işi çalışma zamanında başka bir işi çağıramaz; bu nedenle işleme mantığınız önceki iterasyonun çıktısına bağlı olan yinelemeli yürütmeler gerektirdiğinde, istemci tarafındaki main yöntemiyle birden çok işi koordine etmeniz gerekir. Bu örnekte yineleme döngüsünü kontrol etmek için bir sayaç kullanılır.

Nasıl çalışır?

Program iki tür işi sırayla çalıştırır:

Init job (InitMapper): Çıktı tablosuna başlangıç sayaç değerini (2) yazar.
Decrease jobs (DecreaseMapper): Döngü halinde çalışır. Her yinelemede bir kaynak tablosundan önceki çıktıyı okur, değeri 1 azaltır ve bir sayaç ayarlar. Sayaç 0 değerine ulaştığında döngü sonlanır.

İşler arası veri aktarımı iki mekanizma üzerinden gerçekleşir:

Kaynak tablosu (multijobs_res_table): İşler arasında sayısal değeri taşır.
Sayaç (multijobs.value): main yöntemine yinelemeye devam edilip edilmeyeceğini bildirir.

main yöntemi tek koordinatördür: İşleri sırayla gönderir ve her adımdan sonra tamamlanmayı denetler.

Ön koşullar

Başlamadan önce aşağıdakileri yaptığınızdan emin olun:

Başlarken bölümünde açıklanan ortam kurulumunu tamamlayın
mapreduce-examples.jar paketini derleyin ve MaxCompute istemcisinin kurulu olduğu yol altındaki bin\data\resources dizinine yerleştirin

Tabloları ve kaynakları hazırlayın

Test tablolarını oluşturun:

CREATE TABLE mr_empty (key STRING, value STRING);
CREATE TABLE mr_multijobs_out (value BIGINT);

İş tarafından kullanılan kaynakları kaydedin:
```
add table mr_multijobs_out as multijobs_res_table -f;

-- Omit -f when adding the JAR for the first time.
add jar data\resources\mapreduce-examples.jar -f;
```
mr_multijobs_out tablosu multijobs_res_table olarak kaydedilir; böylece DecreaseMapper önceki işin çıktısını bir kaynak tablosu olarak okuyabilir. JAR dosyası da kaydedilir; bu sayede MaxCompute istemcisi çalışma zamanında ilgili sınıfa erişebilir.

MultiJobs çalıştırın

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

jar -resources mapreduce-examples.jar,multijobs_res_table -classpath data\resources\mapreduce-examples.jar \
    com.aliyun.odps.mapred.open.example.MultiJobs mr_multijobs_out;

Komut parametreleri:

Parametre	Değer	Açıklama
`-resources`	`mapreduce-examples.jar,multijobs_res_table`	İş için kullanılabilen kaynaklar: JAR paketi ve önceki işin çıktısını taşıyan kaynak tablosu
`-classpath`	`data\resources\mapreduce-examples.jar`	İstemcinin ana sınıfı bulmak için kullandığı yerel JAR yolu
Ana sınıf	`com.aliyun.odps.mapred.open.example.MultiJobs`	Programın giriş noktası
Bağımsız değişken	`mr_multijobs_out`	`main` yöntemine iletilen çıktı tablosu adı

Beklenen sonuç

İş tamamlandığında mr_multijobs_out tablosunda tek bir kayıt bulunur:

+------------+
| value      |
+------------+
| 0          |
+------------+

Örnek kod

Project Object Model (POM) bağımlılık yapılandırması için Başlarken kılavuzundaki Önlemler bölümüne bakın.

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.RunningJob;
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;
/**
     * MultiJobs
     *
     * Running multiple job
     *
     **/
public class MultiJobs {
    public static class InitMapper extends MapperBase {
        @Override
            public void setup(TaskContext context) throws IOException {
            Record record = context.createOutputRecord();
            long v = context.getJobConf().getLong("multijobs.value", 2);
            record.set(0, v);
            context.write(record);
        }
    }
    public static class DecreaseMapper extends MapperBase {
        @Override
            public void cleanup(TaskContext context) throws IOException {
            /** Obtain the variable values that are defined in the main function from JobConf. */
            long expect = context.getJobConf().getLong("multijobs.expect.value", -1);
            long v = -1;
            int count = 0;
            /** Read the data from the output table of the previous job. */
            Iterator<Record> iter = context.readResourceTable("multijobs_res_table");
            while (iter.hasNext()) {
                Record r = iter.next();
                v = (Long) r.get(0);
                if (expect != v) {
                    throw new IOException("expect: " + expect + ", but: " + v);
                }
                count++;
            }
            if (count != 1) {
                throw new IOException("res_table should have 1 record, but: " + count);
            }
            Record record = context.createOutputRecord();
            v--;
            record.set(0, v);
            context.write(record);
            /** Set the counter. The counter value can be obtained in the main function after the job is completed. */
            context.getCounter("multijobs", "value").setValue(v);
        }
    }
    public static void main(String[] args) throws Exception {
        if (args.length != 1) {
            System.err.println("Usage: TestMultiJobs <table>");
            System.exit(1);
        }
        String tbl = args[0];
        long iterCount = 2;
        System.err.println("Start to run init job.");
        JobConf initJob = new JobConf();
        initJob.setLong("multijobs.value", iterCount);
        initJob.setMapperClass(InitMapper.class);
        InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), initJob);
        OutputUtils.addTable(TableInfo.builder().tableName(tbl).build(), initJob);
        initJob.setMapOutputKeySchema(SchemaUtils.fromString("key:string"));
        initJob.setMapOutputValueSchema(SchemaUtils.fromString("value:string"));
        /** Explicitly set the number of reducers to 0 for map-only jobs. */
        initJob.setNumReduceTasks(0);
        JobClient.runJob(initJob);
        while (true) {
            System.err.println("Start to run iter job, count: " + iterCount);
            JobConf decJob = new JobConf();
            decJob.setLong("multijobs.expect.value", iterCount);
            decJob.setMapperClass(DecreaseMapper.class);
            InputUtils.addTable(TableInfo.builder().tableName("mr_empty").build(), decJob);
            OutputUtils.addTable(TableInfo.builder().tableName(tbl).build(), decJob);
            /** Explicitly set the number of reducers to 0 for map-only jobs. */
            decJob.setNumReduceTasks(0);
            RunningJob rJob = JobClient.runJob(decJob);
            iterCount--;
            /** If the specified number of iterations is reached, exit the loop. */
            if (rJob.getCounters().findCounter("multijobs", "value").getValue() == 0) {
                break;
            }
        }
        if (iterCount != 0) {
            throw new IOException("Job failed.");
        }
    }
}

Kod incelemesi

InitMapper

setup() içinde tek sefer çalışır. JobConf üzerinden başlangıç sayaç değerini (multijobs.value, varsayılan 2) okur ve çıktı tablosuna yazar. Bu yalnızca map aşamasından oluşan bir iştir; setNumReduceTasks(0) çağrısı çerçevenin reducer başlatmasını engeller.

DecreaseMapper

Her yinelemenin sonunda cleanup() içinde çalışır. Adımları şunlardır:

JobConf üzerinden beklenen değeri (multijobs.expect.value) okur.
context.readResourceTable() kullanarak multijobs_res_table içinden tek kaydı okur. Kayıt sayısı tam olarak 1 değilse veya değer beklenenle eşleşmiyorsa IOException fırlatır.
Değeri 1 azaltır, çıktı tablosuna yazar ve sayacı (multijobs.value) yeni değerle günceller.

main yöntemi

İş sırasını yönetir:

Init işini gönderir ve tamamlanmasını bekler.
Decrease işlerini teker teker gönderen bir döngüye girer; her işin bitmesini bekledikten sonra sonraki adıma geçer.
Her decrease işinden sonra multijobs.value sayacını denetler. Değer 0 ise döngüden çıkar.
Döngü sonrası iterCount değerinin 0 olduğunu doğrular; aksi durumda IOException fırlatarak iş hatasını bildirir.