Data Engineering11 Mei 2026

Building a Self-Healing MongoDB-to-BigQuery ETL Engine in Spring Batch

Bagaimana saya merekayasa pipeline ETL yang sangat tangguh dan adaptif terhadap evolusi skema menggunakan Spring Batch untuk mereplikasi dokumen MongoDB bebas-skema ke BigQuery dalam mode Full dan Incremental.

#Spring Boot#Spring Batch#MongoDB#BigQuery#ETL#Schema Evolution#Self-Healing

Ringkasan Eksekutif

Mereplikasi data transaksional terdistribusi dari MongoDB (database NoSQL berorientasi dokumen yang bebas skema) ke Google Cloud BigQuery (data warehouse relasional berskema ketat) memiliki tantangan unik. Kendala terbesar dalam produksi adalah evolusi skema yang tidak terprediksi (dynamic schemas) serta konflik tipe data (polymorphic fields—seperti kolom userId yang pada beberapa dokumen lama bertipe integer/numeric, tetapi pada dokumen baru bertipe string atau nested object).

Pada sistem warisan (legacy), setiap kali tim produk merilis fitur baru dengan field baru atau tipe data yang berbeda di MongoDB, pipeline ETL akan macet seketika karena BigQuery menolak skema yang tidak cocok. Hal ini menuntut intervensi manual dari data engineer untuk melakukan pemetaan ulang kolom.

Untuk menyelesaikan masalah ini secara permanen, saya merancang dan membangun mesin ETL asinkron kustom berbasis Spring Boot & Spring Batch. Mesin ini mampu mereplikasi data secara efisien melalui integrasi GCS (Google Cloud Storage) staging, mendeteksi dan memperluas skema BigQuery secara otomatis (Schema Evolution), serta memulihkan dirinya sendiri dari konflik tipe data (Polymorphic Self-Healing) tanpa mematikan pipeline berjalan.

Arsitektur Pipeline Parallel Multi-Collection

Mesin ETL ini memanfaatkan keunggulan pemrosesan asinkron dan multi-threading pada Spring Batch untuk menyinkronkan beberapa koleksi MongoDB sekaligus secara paralel. Aliran data untuk setiap koleksi dibagi menjadi tiga langkah berurutan (sequential steps):

Read & Buffer Step (mongo_read_{collection}): Membaca data secara chunk-based dari MongoDB dan menulisnya ke dalam berkas berformat JSONL (JSON Lines) lokal secara berurutan.
Upload Step (gcs_upload_{collection}): Mengunggah berkas JSONL lokal ke Google Cloud Storage (GCS) staging bucket.
Load Step (bq_load_{collection}): Menginisiasi pemanggilan API Google Cloud BigQuery Load Job untuk memuat file dari GCS ke tabel produksi, sambil menjalankan validasi skema dan resolusi konflik tipe data.

Mekanisme ini digambarkan dalam diagram alur paralel berikut:

graph TD
    subgraph Job [Spring Batch: mongoGcsTransferJob]
        Split[SplitState: Parallel Execution] -->|Thread 1| Flow_A[Flow: collection_A]
        Split -->|Thread 2| Flow_B[Flow: collection_B]
        
        subgraph Flow_A [Flow: Collection A]
            Read_A[Step 1: mongo_read_A <br> Read Mongo -> Write local JSONL] -->|Success| Upload_A[Step 2: gcs_upload_A <br> Upload JSONL to GCS]
            Upload_A -->|Success| Load_A[Step 3: bq_load_A <br> Reconcile Schema -> BQ Load Job]
        end

        subgraph Flow_B [Flow: Collection B]
            Read_B[Step 1: mongo_read_B <br> Read Mongo -> Write local JSONL] -->|Success| Upload_B[Step 2: gcs_upload_B <br> Upload JSONL to GCS]
            Upload_B -->|Success| Load_B[Step 3: bq_load_B <br> Reconcile Schema -> BQ Load Job]
        end
    end

    style Split fill:#f9f,stroke:#333,stroke-width:1px
    style Flow_A fill:#f5f5f5,stroke:#aaa,stroke-width:1px
    style Flow_B fill:#f5f5f5,stroke:#aaa,stroke-width:1px

Mekanisme Pendeteksian Watermark Dua Dimensi (Chronological Slicing)

Untuk menjalankan Incremental Mode secara efisien tanpa kehilangan data akibat keterlambatan sinkronisasi, mesin ETL melakukan penarikan watermark secara dinamis dari BigQuery sebelum membaca database MongoDB.

Jika kueri inkremental tradisional hanya menyaring berdasarkan waktu pembaruan (updated_at > last_watermark), risiko kehilangan data sangat besar jika ada dua dokumen atau lebih yang memiliki timestamp pembaruan yang tepat sama milidetiknya, dan sebagian data belum terbaca pada eksekusi sebelumnya.

Untuk mengatasinya, saya menerapkan Watermark Dua Dimensi yang menggabungkan stempel waktu (updated_at) dan pengenal unik dokumen (_id):

-- Query Dynamic Watermark di BigQuery
SELECT 
  FORMAT_TIMESTAMP('%FT%H:%M:%E6SZ', updated_at, 'UTC') AS max_watermark, 
  CAST(_id AS STRING) AS last_id 
FROM 
  `project_id.dataset_name.table_name` 
WHERE 
  updated_at IS NOT NULL 
ORDER BY 
  updated_at DESC, 
  CAST(_id AS STRING) DESC 
LIMIT 1;

Berdasarkan hasil kueri di atas, MongoGcsItemReader akan menyusun kueri pencarian MongoDB secara dinamis menggunakan operator logika OR:

Dokumen yang memiliki updated_at lebih besar dari last_watermark_timestamp.
ATAU dokumen yang memiliki updated_at sama persis dengan last_watermark_timestamp tetapi memiliki _id lebih besar dari last_watermark_id (menggunakan sortasi leksikografis ID).

Ini menjamin konsistensi data 100% (exactly-once ingestion contract) tanpa adanya data terlewat.

Solusi Self-Healing Bentrok Tipe Data (Polymorphic Conflict Resolution)

Bila terjadi bentrok tipe data antara tabel BigQuery yang ada dengan skema dokumen MongoDB baru, mesin ETL akan bertindak berdasarkan konfigurasi kebijakan resolusi konflik skema (schemaConflictPolicy):

BqPolymorphicDetector (Tahap Sampling): Sebelum memproses file JSONL, detektor memindai 1000 record pertama dari file GCS. Jika mendeteksi ada suatu kolom yang memiliki lebih dari satu jenis tipe data aktif di dalamnya (misal: kolom payload terdeteksi diisi teks STRING dan di dokumen lain diisi objek RECORD), kolom tersebut langsung ditandai sebagai polymorphic.
Schema Coercion: Kolom polymorphic tersebut secara otomatis dipaksa (coerced) untuk didefinisikan sebagai tipe data STRING di BigQuery. Nilai JSON yang kompleks akan diserialisasikan menjadi JSON String biasa agar proses load tidak mengalami kegagalan.
Companion Field Resolution (__str Suffix): Jika konflik tipe data baru terdeteksi saat membandingkan skema inferred (hasil ekstraksi data baru) dengan skema tabel BigQuery yang sudah ada, mesin ETL tidak akan menghentikan proses. Melainkan, mesin akan secara dinamis membuat kolom baru dengan nama akhiran __str (misal: jika kolom asli userId bertipe INTEGER, maka kolom baru userId__str bertipe STRING akan ditambahkan via DDL ALTER TABLE). Data baru yang bertipe bentrok tersebut dialihkan ke kolom companion ini.

Berikut adalah cuplikan kode komponen utama BqSchemaReconciler yang menangani pembentukan kolom companion saat terjadi konflik tipe:

private void handleConflictByPolicy(
        Map<String, Field> merged,
        Map<String, String> canonicalNameByLowerCase,
        Field existingField,
        Field inferredField,
        String fieldPath,
        String schemaConflictPolicy) {
    
    if ("fail-fast".equalsIgnoreCase(schemaConflictPolicy)) {
        throw new IllegalStateException(String.format(
                "Schema conflict on %s: existingType=%s, inferredType=%s",
                fieldPath, existingField.getType(), inferredField.getType()));
    }

    // Kebijakan "coerce-to-string" -> membuat companion field bertipe STRING
    String companionName = existingField.getName() + "__str";
    String existingCompanionName = canonicalNameByLowerCase.get(companionName.toLowerCase());
    
    if (existingCompanionName == null) {
        Field companionField = Field.newBuilder(companionName, LegacySQLTypeName.STRING)
                .setMode(Field.Mode.NULLABLE)
                .build();
        merged.put(companionName, companionField);
        canonicalNameByLowerCase.put(companionName.toLowerCase(), companionName);
        log.warn("Schema type conflict on {}: existingType={}, inferredType={}. Added companion field '{}' (STRING)",
                fieldPath, existingField.getType(), inferredField.getType(), companionName);
    }
}

Implementasi Orkestrasi Job Spring Batch

Berikut adalah cuplikan kelas konfigurasi pekerjaan batch utama BatchConfigurationMongoGcs yang mengatur paralelisme pemrosesan multi-koleksi menggunakan SplitState dan memadukan pembaca inkremental serta pemuatan BigQuery:

@Configuration
public class BatchConfigurationMongoGcs {

    private final JobRepository jobRepository;
    private final PlatformTransactionManager transactionManager;
    private final MongoGcsTableResolverService resolverService;

    // Executor untuk pemrosesan paralel koleksi MongoDB
    @Bean("mongoGcsTaskExecutor")
    @JobScope
    public TaskExecutor mongoGcsTaskExecutor(@Value("#{jobParameters['parallelismLevel']}") String parallelism) {
        int coreSize = parallelism != null ? Integer.parseInt(parallelism) : 2;
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(coreSize);
        executor.setMaxPoolSize(10);
        executor.setThreadNamePrefix("mongo-gcs-");
        executor.initialize();
        return executor;
    }

    @Bean("mongoGcsTransferJob")
    public Job mongoGcsTransferJob(List<MongoGcsCollectionConfig> configs, TaskExecutor executor) {
        JobBuilder jobBuilder = new JobBuilder("mongoGcsTransferJob", jobRepository)
                .incrementer(new RunIdIncrementer());

        // Hasilkan flows terpisah untuk setiap koleksi
        List<Flow> flows = configs.stream()
                .filter(MongoGcsCollectionConfig::getEnabled)
                .map(this::buildCollectionFlow)
                .toList();

        // Bagi aliran kerja ke beberapa thread (Paralel)
        return jobBuilder
                .start(flows.get(0))
                .split(executor)
                .add(flows.subList(1, flows.size()).toArray(new Flow[0]))
                .end()
                .build();
    }

    private Flow buildCollectionFlow(MongoGcsCollectionConfig config) {
        String name = config.getName();
        Step readStep = new StepBuilder("mongo_read_" + name, jobRepository)
                .<Map<String, Object>, Map<String, Object>>chunk(config.getChunkSize(), transactionManager)
                .reader(resolverService.createReader(config))
                .writer(resolverService.createWriter(config))
                .build();

        Step gcsUploadStep = buildGcsUploadStep(config);
        Step bqLoadStep = buildBqLoadStep(config);

        return new FlowBuilder<SimpleFlow>("flow_" + name)
                .start(readStep)
                .next(gcsUploadStep)
                .next(bqLoadStep)
                .build();
    }
}

Dampak Bisnis & Hasil Kerja

Solusi ETL mandiri yang dikembangkan ini membawa dampak operasional dan efisiensi penanganan data yang signifikan bagi platform data perusahaan:

Zero Ingestion Failures: Mekanisme polymorphic self-healing memangkas angka kegagalan ingesti BigQuery dari yang sebelumnya terjadi beberapa kali setiap bulan menjadi 0%.
Otomatisasi Penuh Evolusi Skema: Skema tabel produksi BigQuery diperbarui secara dinamis tanpa downtime. Menghemat waktu pengembangan tim data engineer hingga 15+ jam kerja per minggu karena tidak perlu melakukan manipulasi DDL manual.
Peningkatan Efisiensi Pipeline MoEngage: Dengan data terstruktur BigQuery yang diperbarui tepat waktu secara paralel, sinkronisasi segmen pengguna untuk kampanye pemasaran via CRM MoEngage berjalan tanpa hambatan, memotong waktu tunggu tim operasional pemasaran hingga 24 jam.
Ingesti Cepat: Paralelisme Spring Batch mempercepat laju ingesti data kumulatif dari puluhan koleksi MongoDB secara simultan, menuntaskan pemrosesan jutaan baris data dalam waktu kurang dari 15 menit.