Memproses satu juta baris dalam satu tarikan terasa efisien sampai tiga hal terjadi: memori membengkak karena seluruh data termuat, tidak ada laporan kemajuan selama berjam-jam, dan satu kegagalan di baris ke-999.000 memaksa pengulangan dari awal. Batch processing menjawab ketiganya sekaligus: data dipotong menjadi satuan kerja kecil, tiap satuan dilaporkan, dan tiap satuan yang selesai dicatat ke checkpoint sehingga kegagalan hanya mengulang potongan terakhir. Pola ini menjadi inti dari pipeline ETL malam-harian di perusahaan — dan versi kecilnya bisa kita bangun dengan NIO murni. Chapter ini merakit pipeline lengkap: baca data transaksi, proses per batch, tulis checkpoint setelah tiap batch, lalu buktikan resume bekerja setelah kegagalan disengaja.

Batch sebagai Satuan Kerja #

Dua keputusan desain menentukan bentuk program. Pertama, ukuran batch membatasi memori dan menentukan granularitas resume — lima baris untuk demo, ribuan untuk produksi. Kedua, checkpoint ditulis setelah batch selesai, bukan sebelum: mencatat lebih dulu berarti mengaku pekerjaan yang belum dilakukan, sehingga kegagalan di tengah batch justru kehilangan data. Makna kedua pola ini dirangkum tabel berikut:

Saat checkpoint ditulisJaminan bila proses gagalKonsekuensi
Sebelum batch dieksekusidata bisa terlewat (at-most-once)baris hilang senyap
Setelah batch dieksekusidata bisa terproses dua kali (at-least-once)butuh langkah pemulihan

Untuk agregasi penjumlahan, pemrosesan ulang tidak mengubah hasil. Untuk efek samping seperti pengiriman surel, pilih at-least-once lalu tandai item yang sudah diproses di sisi tujuan.

Pipeline dengan Checkpoint dan Resume #

Program berikut membaca dua belas baris transaksi dari transaksi.txt, memprosesnya per lima baris, dan menyimpan posisi terakhir yang sah di checkpoint.txt. Kegagalan disimulasikan terjadi pada batch ke-3 dari proses pertama; fase kedua — seolah program dijalankan ulang setelah insiden — membaca checkpoint dan melanjutkan tanpa mengulang batch yang sudah aman:

java ProsesBatch.java 
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;

public class ProsesBatch {
    static final int UKURAN_BATCH = 5;
    static final Path CHECKPOINT = Path.of("checkpoint.txt");

    public static void main(String[] args) throws IOException {
        List<String> baris = Files.readAllLines(Path.of("transaksi.txt"),
                StandardCharsets.UTF_8);
        System.out.println("total baris masukan: " + baris.size());

        // Fase 1: kegagalan disimulasikan menghentikan proses di batch ke-3.
        System.out.println("== proses pertama ==");
        jalankan(baris, false);
        System.out.println("posisi aman : " + bacaCheckpoint() + " dari " + baris.size());

        // Fase 2: program dijalankan ulang; checkpoint menentukan titik lanjut.
        System.out.println("== proses ulang ==");
        jalankan(baris, true);
        System.out.println("posisi aman : " + bacaCheckpoint() + " dari " + baris.size());

        long total = baris.stream().mapToLong(ProsesBatch::nilai).sum();
        System.out.println("total nilai : " + total);
    }

    static void jalankan(List<String> baris, boolean faseUlang) throws IOException {
        int posisi = bacaCheckpoint();          // titik lanjut terakhir yang sah
        if (posisi > 0) {
            System.out.println("melanjutkan dari baris " + (posisi + 1));
        }
        while (posisi < baris.size()) {
            int batchKe = posisi / UKURAN_BATCH + 1;
            int awal = posisi;
            int akhir = Math.min(awal + UKURAN_BATCH, baris.size());

            // Kegagalan disimulasikan hanya pada proses pertama, batch ke-3.
            if (!faseUlang && batchKe == 3) {
                System.out.println("batch " + batchKe + ": baris " + (awal + 1)
                        + "-" + akhir + " -- GAGAL: kegagalan disimulasikan");
                return;                          // checkpoint batch ini tidak ditulis
            }

            long nilaiBatch = 0;
            for (int i = awal; i < akhir; i++) {
                nilaiBatch += nilai(baris.get(i));
            }
            posisi = akhir;
            tulisCheckpoint(posisi);             // ditulis SETELAH batch selesai
            System.out.println("batch " + batchKe + ": baris " + (awal + 1) + "-" + akhir
                    + " nilai=" + nilaiBatch + " (checkpoint: " + posisi + ")");
        }
    }

    static long nilai(String baris) {
        return Long.parseLong(baris.split(",")[2]);
    }

    static int bacaCheckpoint() throws IOException {
        if (!Files.exists(CHECKPOINT)) {
            return 0;
        }
        return Integer.parseInt(Files.readString(CHECKPOINT).trim());
    }

    static void tulisCheckpoint(int posisi) throws IOException {
        Files.writeString(CHECKPOINT, Integer.toString(posisi));
    }
}

Output yang Dihasilkan:

text 
total baris masukan: 12
== proses pertama ==
batch 1: baris 1-5 nilai=29500 (checkpoint: 5)
batch 2: baris 6-10 nilai=20500 (checkpoint: 10)
batch 3: baris 11-12 -- GAGAL: kegagalan disimulasikan
posisi aman : 10 dari 12
== proses ulang ==
melanjutkan dari baris 11
batch 3: baris 11-12 nilai=12500 (checkpoint: 12)
posisi aman : 12 dari 12
total nilai : 62500

Mari kita telaah bagian-bagian kuncinya:

  1. Baris int posisi = bacaCheckpoint(): setiap pemanggilan jalankan memulai dari posisi tersimpan — program ulang tidak tahu apa pun tentang kegagalan sebelumnya; checkpoint adalah satu-satunya memori lintas eksekusi.
  2. Baris int batchKe = posisi / UKURAN_BATCH + 1: nomor batch dihitung dari posisi, bukan penghitung lokal, sehingga penomoran tetap konsisten setelah resume — batch ke-3 tetap bernomor 3.
  3. Baris return pada cabang GAGAL: kegagalan keluar sebelum tulisCheckpoint, sehingga posisi aman berhenti di angka 10 — dua batch pertama tidak perlu diulang.
  4. Baris tulisCheckpoint(posisi) setelah loop nilai: penulisan dilakukan setelah seluruh baris batch terproses — pola at-least-once dari tabel di atas.
  5. Baris melanjutkan dari baris 11: fase kedua melompat langsung ke sisa data; keluaran “posisi aman: 12 dari 12” membuktikan seluruh baris tersentuh tepat satu kali dengan total yang benar.

Untuk berkas yang tidak muat di memori, ganti readAllLines dengan Files.lines(path, UTF_8) — stream yang dibaca bertahap — dan simpan checkpoint berupa nomor baris, lalu skip() nomor tersebut saat resume. Struktur loop-nya tidak berubah.

Pipeline kita kini tahan kegagalan untuk satu berkas. Otomasi berikutnya menyapu banyak berkas sekaligus — memindahkan, mengganti nama, dan merapikan pohon direktori dengan pattern matching.

Referensi #

Berkas Sumber

Berkas pendukung halaman ini dapat diunduh dan dijalankan langsung sesuai penjelasan di atas.

ProsesBatch.java

Unduh berkas
java 
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;

public class ProsesBatch {
    static final int UKURAN_BATCH = 5;
    static final Path CHECKPOINT = Path.of("checkpoint.txt");

    public static void main(String[] args) throws IOException {
        List<String> baris = Files.readAllLines(Path.of("transaksi.txt"),
                StandardCharsets.UTF_8);
        System.out.println("total baris masukan: " + baris.size());

        // Fase 1: kegagalan disimulasikan menghentikan proses di batch ke-3.
        System.out.println("== proses pertama ==");
        jalankan(baris, false);
        System.out.println("posisi aman : " + bacaCheckpoint() + " dari " + baris.size());

        // Fase 2: program dijalankan ulang; checkpoint menentukan titik lanjut.
        System.out.println("== proses ulang ==");
        jalankan(baris, true);
        System.out.println("posisi aman : " + bacaCheckpoint() + " dari " + baris.size());

        long total = baris.stream().mapToLong(ProsesBatch::nilai).sum();
        System.out.println("total nilai : " + total);
    }

    static void jalankan(List<String> baris, boolean faseUlang) throws IOException {
        int posisi = bacaCheckpoint();          // titik lanjut terakhir yang sah
        if (posisi > 0) {
            System.out.println("melanjutkan dari baris " + (posisi + 1));
        }
        while (posisi < baris.size()) {
            int batchKe = posisi / UKURAN_BATCH + 1;
            int awal = posisi;
            int akhir = Math.min(awal + UKURAN_BATCH, baris.size());

            // Kegagalan disimulasikan hanya pada proses pertama, batch ke-3.
            if (!faseUlang && batchKe == 3) {
                System.out.println("batch " + batchKe + ": baris " + (awal + 1)
                        + "-" + akhir + " -- GAGAL: kegagalan disimulasikan");
                return;                          // checkpoint batch ini tidak ditulis
            }

            long nilaiBatch = 0;
            for (int i = awal; i < akhir; i++) {
                nilaiBatch += nilai(baris.get(i));
            }
            posisi = akhir;
            tulisCheckpoint(posisi);             // ditulis SETELAH batch selesai
            System.out.println("batch " + batchKe + ": baris " + (awal + 1) + "-" + akhir
                    + " nilai=" + nilaiBatch + " (checkpoint: " + posisi + ")");
        }
    }

    static long nilai(String baris) {
        return Long.parseLong(baris.split(",")[2]);
    }

    static int bacaCheckpoint() throws IOException {
        if (!Files.exists(CHECKPOINT)) {
            return 0;
        }
        return Integer.parseInt(Files.readString(CHECKPOINT).trim());
    }

    static void tulisCheckpoint(int posisi) throws IOException {
        Files.writeString(CHECKPOINT, Integer.toString(posisi));
    }
}