SchedulerX:Pekerjaan Java

Execution modes

Pekerjaan Java mendukung mode eksekusi berikut:

• Standalone: Pekerjaan dijalankan pada mesin acak dalam groupId yang sama.

• Broadcast: Pekerjaan dijalankan secara konkuren pada semua mesin dalam groupId yang sama.

• Visual MapReduce: Ini adalah model tugas MapReduce yang memerlukan Professional Edition. Mode ini mendukung hingga 1.000 subtask. Anda dapat mengkueri catatan eksekusi terperinci, log operasional, dan stack subtask menggunakan kata kunci bisnis.

• MapReduce: Ini adalah tugas model MapReduce reguler yang mendukung pemrosesan paralel banyak subtask. Anda hanya dapat mengkueri informasi ringkasan mengenai eksekusi subtask. Pilih mode ini ketika jumlah subtask kurang dari satu juta.

• Sharded execution: Mode ini menggunakan sharding statis dan batching dinamis untuk memproses data besar.

Implementasikan JavaProcessor untuk mode Standalone dan Broadcast. Implementasikan MapJobProcessor untuk mode Visual MapReduce, MapReduce, dan Sharded execution.

Jalur kelas processor adalah nama lengkap kelas implementasi. Contohnya, com.apache.armon.test.schedulerx.processor.MySimpleJob.

• Jika Anda tidak mengunggah paket JAR, SchedulerX mencari kelas implementasi processor dalam classpath aplikasi Anda. Oleh karena itu, Anda harus mengompilasi ulang dan menerbitkan aplikasi setelah setiap modifikasi.

• Jika Anda mengunggah paket JAR, SchedulerX melakukan hot-reload terhadap paket JAR dan processor tersebut. Anda tidak perlu menerbitkan ulang aplikasi.

Programming models

Pekerjaan Java mendukung dua model pemrograman: JavaProcessor dan MapJobProcessor.

• JavaProcessor

  • Opsional: public void preProcess(JobContext context) throws Exception

  • Wajib: public ProcessResult process(JobContext context) throws Exception

  • Opsional: public void postProcess(JobContext context)

  • Opsional: public void kill(JobContext context)

• MapJobProcessor

  • Wajib: public ProcessResult process(JobContext context) throws Exception

  • Opsional: public void postProcess(JobContext context)

  • Opsional: public void kill(JobContext context)

  • Wajib: public ProcessResult map(List<? extends Object> taskList, String taskName)

ProcessResult

Setiap proses harus mengembalikan objek ProcessResult. Objek ini menunjukkan status eksekusi tugas, hasil, dan pesan error.

• Tugas berjalan berhasil dan mengembalikan new ProcessResult(true).

• Tugas gagal jika mengembalikan new ProcessResult(false, ErrorMsg) atau melemparkan exception.

• Eksekusi tugas yang berhasil mengembalikan return new ProcessResult(true, result), dengan result berupa string yang tidak boleh melebihi 1.000 byte.

Contoh pekerjaan HelloSchedulerx2.0

@Component
public class MyProcessor1 extends JavaProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        //TODO
        System.out.println("Hello, schedulerx2.0!");
        return new ProcessResult(true);
    }
}            

Contoh pekerjaan yang mendukung fitur kill

@Component
public class MyProcessor2 extends JavaProcessor {
    private volatile boolean stop = false;

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        int N = 10000;
        while (!stop && N >= 0) {
            //TODO
            N--;
        }
        return new ProcessResult(true);
    }

    @Override
    public void kill(JobContext context) {
        stop = true;
    }

    @Override
    public void preProcess(JobContext context) {
        stop = false;  // Jika pekerjaan diluncurkan menggunakan Spring dan bean-nya singleton, gunakan preProcess untuk mengatur ulang flag.
    }
}          

Contoh pekerjaan untuk pemrosesan batch menggunakan model Map

/**
 * Pemrosesan batch terdistribusi untuk tabel non-partisi.
 * 1. Tugas root mengkueri tabel untuk mendapatkan minId dan maxId.
 * 2. Buat PageTask dan distribusikan menggunakan metode map.
 * 3. Jika level berikutnya menerima PageTask, proses datanya.
 *
 */
@Component
public class ScanSingleTableJobProcessor extends MapJobProcessor {
    private static final int pageSize = 100;

    static class PageTask {
        private int startId;
        private int endId;
        public PageTask(int startId, int endId) {
             this.startId = startId;
             this.endId = endId;
        }
        public int getStartId() {
              return startId;
        }
        public int getEndId() {
              return endId;
        }
    }

    @Override
    public ProcessResult process(JobContext context) {
        String taskName = context.getTaskName();
        Object task = context.getTask();
        if (isRootTask(context)) {
            System.out.println("start root task");
            Pair<Integer, Integer> idPair = queryMinAndMaxId();
            int minId = idPair.getFirst();
            int maxId = idPair.getSecond();
            List<PageTask> taskList = Lists.newArrayList();
            int step = (int) ((maxId - minId) / pageSize); // Hitung jumlah halaman.
            for (int i = minId; i < maxId; i+=step) {
                taskList.add(new PageTask(i, (i+step > maxId ? maxId : i+step)));
            }
            return map(taskList, "Level1Dispatch"); // Proses memanggil metode map untuk mendistribusikan subtask.
        } else if (taskName.equals("Level1Dispatch")) {
            PageTask record = (PageTask)task;
            long startId = record.getStartId();
            long endId = record.getEndId();
            //TODO
            return new ProcessResult(true);
        }

        return new ProcessResult(true);
    }

    @Override
    public void postProcess(JobContext context) {
        //TODO
        System.out.println("Semua tugas selesai.");
    }

    private Pair<Integer, Integer> queryMinAndMaxId() {
        //TODO select min(id),max(id) from xxx
        return null;
    }

}