Data Architecture

Membangun Incremental Watermark-Driven Reverse ETL dengan Spring Batch

Bagaimana saya merancang dan merekayasa aplikasi Java Reverse ETL kustom menggunakan Spring Batch untuk menyinkronkan data analitik olahan dari BigQuery kembali ke Cloud SQL PostgreSQL dengan aman dan efisien.

#Spring Boot#Spring Batch#BigQuery#PostgreSQL#Cloud SQL#Reverse ETL#Watermark

Teknologi Implementasi: Java 17 / 21, Spring Boot 3.x, Spring Batch 5.x, GCP BigQuery, Cloud SQL (PostgreSQL).

Ringkasan Eksekutif

Dalam arsitektur data modern, pipeline data tidak lagi hanya berupa jalan satu arah menuju data warehouse. Wawasan analitik hasil olahan dari data warehouse sering kali perlu disinkronkan kembali ke database operasional di hilir untuk mendukung kebutuhan transaksional, personalisasi layanan, dan operasional secara langsung. Proses ini dikenal sebagai Reverse ETL. Sebagai studi kasus, saya mengambil contoh arsitektur yang digunakan di AstraPay, di mana data analitik hasil olahan dari Google Cloud BigQuery disinkronkan kembali ke database operasional Cloud SQL PostgreSQL untuk meningkatkan efisiensi pemrosesan data.

Menyinkronkan data dengan cara menulis ulang seluruh tabel (Full Load) sangat tidak efisien. Hal ini menyebabkan kemacetan penulisan (write bottlenecks) pada database transaksional operasional, menguras sumber daya CPU/jaringan, serta memicu peningkatan biaya pemindaian query (query scan fees) di data warehouse.

Untuk mengatasi kendala ini, saya merancang dan mengimplementasikan Reverse ETL engine mandiri menggunakan Spring Boot dan Spring Batch—sebuah framework Java skala industri yang dirancang khusus untuk pemrosesan data volume besar dengan fitur bawaan seperti pemrosesan berbasis chunk (chunk-oriented), manajemen transaksi, dan ketahanan terhadap kegagalan (fault tolerance).

Secara sederhana, Spring Batch membagi tugas besar (Job) menjadi beberapa tahapan (Step). Dalam Step berbasis chunk, data dibaca satu per satu dari sumber, dikumpulkan dalam memori (buffer) hingga batas ukuran tertentu (misalnya 1000 baris), lalu ditulis sekaligus secara massal (bulk) ke database tujuan dalam satu transaksi tunggal. Model pemrosesan ini dapat digambarkan sebagai berikut:

graph TD
    subgraph Job [Spring Batch Job]
        subgraph Step [Step: Chunk-Oriented Processing]
            Reader[ItemReader <br> Membaca data per baris dari BigQuery] -->|1. Baca baris demi baris| Buffer(Chunk Buffer)
            Buffer -->|2. Jika kuota chunk tercapai misal: 1000| Writer[ItemWriter <br> Menulis bulk upsert ke PostgreSQL dalam satu transaksi]
        end
    end

Pola arsitektur ini menggunakan metode Watermark Column Scan untuk mendeteksi perubahan delta data, memungkinkan pipa data beroperasi dalam tiga mode: Full Load, Incremental, dan Merge/Upsert.

Arsitektur Alur & Ingesti Data

Alur sinkronisasi diatur di sekitar mesin metadata pusat yang memeriksa kondisi database target terlebih dahulu sebelum menarik data perubahan (delta) dari data warehouse:

sequenceDiagram
    participant Postgres as Cloud SQL (PostgreSQL)
    participant Batch as Spring Batch Engine (Java)
    participant BQ as GCP BigQuery
    
    Batch->>Postgres: 1. Scan Max Watermark (e.g. SELECT MAX(updated_at))
    Postgres-->>Batch: Return max_watermark value
    Batch->>BQ: 2. Query dynamic delta (WHERE updated_at > max_watermark)
    BQ-->>Batch: Stream matching records (BigQueryItemReader)
    Batch->>Postgres: 3. Chunked upsert/write updates (PostgresItemWriter - Merge Mode)
  1. Ekstraksi Watermark: Sebelum aliran data dimulai, proses batch akan memindai tabel PostgreSQL target untuk mengambil nilai stempel waktu (timestamp) sinkronisasi terakhir.
  2. Ingesti Dinamis (Lazy Ingestion): Berdasarkan nilai watermark yang diperoleh, komponen ItemReader Spring Batch secara dinamis menghasilkan dan mengeksekusi kueri SQL standar yang terfilter ke BigQuery.
  3. Penulisan Tangguh Berbasis Chunk: Baris data yang diambil akan dialirkan dalam unit transaksi (chunk-oriented processing) dan digabungkan (merged) ke database PostgreSQL target menggunakan perintah SQL upsert berkinerja tinggi, sambil terus melacak stempel waktu pemrosesan maksimum.

Implementasi Arsitektur Utama

Catatan: Diagram sekuensial di atas dan cuplikan kode di bawah ini menggambarkan Mode Incremental/Merge (sinkronisasi delta berbasis watermark), yang mewakili kompleksitas arsitektur inti dari mesin ETL ini. Implementasi telah disederhanakan agar mudah dibaca dan dibersihkan dari logika rahasia perusahaan.

1. Resolusi Watermark secara Lazy (Lazy Watermark Resolution)

Kesalahan umum dalam konfigurasi Spring Batch adalah menjalankan kueri metadata pada saat inisialisasi Spring Application Context (pembuatan bean). Hal ini dapat menyebabkan kegagalan sistem saat startup jika database target tidak dapat dihubungi sementara waktu, atau jika data watermark berubah di antara waktu startup aplikasi dan eksekusi job.

Untuk menyelesaikannya, saya merancang Lazy Watermark Supplier. Kueri untuk mengambil stempel waktu maksimum (SELECT MAX(watermark_col)) dibungkus dalam antarmuka Java Supplier<Instant>. Database target hanya akan dikueri ketika langkah (step) batch benar-benar mulai dieksekusi:

import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.jdbc.core.JdbcTemplate;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Map;
import java.util.function.Supplier;

public class JobConfiguration {

    private final JdbcTemplate targetJdbcTemplate;

    public JobConfiguration(JdbcTemplate targetJdbcTemplate) {
        this.targetJdbcTemplate = targetJdbcTemplate;
    }

    public Step transferStep(BigQuery bigQueryClient, String dataset, String table) {
        return new StepBuilder("transferStep", jobRepository)
                .<Map<String, Object>, Map<String, Object>>chunk(1000, transactionManager)
                .reader(createBigQueryReader(bigQueryClient, dataset, table))
                .writer(createPostgresWriter(table))
                .build();
    }

    private ItemReader<Map<String, Object>> createBigQueryReader(
            BigQuery bigQueryClient, String dataset, String table) {
        // Default ke "updated_at" untuk keperluan demonstrasi Mode Incremental/Merge
        String watermarkColumn = "updated_at";
        // Supplier akan menyelesaikan nilai watermark saat job mulai dieksekusi
        Supplier<Instant> watermarkSupplier = () -> resolveLastWatermark(table, watermarkColumn);
        
        return new BigQueryItemReader(
                bigQueryClient, 
                dataset, 
                table, 
                watermarkColumn, 
                watermarkSupplier
        );
    }

    private Instant resolveLastWatermark(String tableName, String watermarkColumn) {
        // Sanitasi pengenal SQL sangat penting untuk mencegah SQL Injection
        String sanitizedTable = sanitizeIdentifier(tableName);
        String sanitizedColumn = sanitizeIdentifier(watermarkColumn);
        String sql = String.format("SELECT MAX(%s) FROM %s", sanitizedColumn, sanitizedTable);
        
        try {
            LocalDateTime maxTimestamp = targetJdbcTemplate.queryForObject(sql, LocalDateTime.class);
            if (maxTimestamp == null) return null;
            return maxTimestamp.atZone(ZoneOffset.UTC).toInstant();
        } catch (Exception e) {
            // Kembali ke null (memicu full load) jika tabel kosong atau terjadi kesalahan
            return null;
        }
    }

    private String sanitizeIdentifier(String name) {
        if (name == null || !name.matches("^[a-zA-Z_][a-zA-Z0-9_]*$")) {
            throw new IllegalArgumentException("Pengenal database tidak aman terdeteksi");
        }
        return name;
    }

    private ItemWriter<Map<String, Object>> createPostgresWriter(String table) {
        return new PostgresItemWriter(targetJdbcTemplate, table);
    }
}

2. Streaming BigQuery Item Reader

Mengambil jutaan baris data sekaligus dari BigQuery dan memuatnya ke memori JVM akan langsung memicu error OutOfMemoryError.

Komponen kustom BigQueryItemReader mengimplementasikan antarmuka ItemReader dan ItemStream dari Spring Batch. Kelas ini memproses kueri BigQuery menggunakan paginasi SQL standar atau iterator streaming bawaan Google (TableResult.iterateAll()). Pembaca ini memetakan setiap baris BigQuery (FieldValueList) ke dalam format Java Map<String, Object> generik dan mencatat stempel waktu watermark maksimum yang ditemui pada setiap chunk data.

import com.google.cloud.bigquery.*;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import java.time.Instant;
import java.util.*;
import java.util.function.Supplier;

public class BigQueryItemReader implements ItemReader<Map<String, Object>>, ItemStream {

    private final BigQuery bigQuery;
    private final String dataset;
    private final String table;
    private final String watermarkColumn;
    private final Supplier<Instant> watermarkSupplier;
    
    private Iterator<FieldValueList> rowIterator;
    private Schema tableSchema;
    private Instant lastWatermark;
    private Instant maxWatermarkSeen;

    public BigQueryItemReader(BigQuery bigQuery, String dataset, String table, 
                              String watermarkColumn, Supplier<Instant> watermarkSupplier) {
        this.bigQuery = bigQuery;
        this.dataset = dataset;
        this.table = table;
        this.watermarkColumn = watermarkColumn;
        this.watermarkSupplier = watermarkSupplier;
    }

    @Override
    public void open(ExecutionContext executionContext) {
        // Resolusi lazy watermark sebelum mengeksekusi kueri
        this.lastWatermark = watermarkSupplier.get();
        this.maxWatermarkSeen = this.lastWatermark;

        Table bqTable = bigQuery.getTable(dataset, table);
        this.tableSchema = bqTable.getDefinition().getSchema();

        String query = buildDynamicQuery();
        QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(query)
                .setUseLegacySql(false)
                .build();

        try {
            TableResult result = bigQuery.query(queryConfig);
            this.rowIterator = result.iterateAll().iterator();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Eksekusi query BigQuery terinterupsi", e);
        }
    }

    private String buildDynamicQuery() {
        String baseQuery = String.format("SELECT * FROM `%s.%s`", dataset, table);
        if (lastWatermark != null && watermarkColumn != null) {
            return String.format("%s WHERE %s > TIMESTAMP('%s')", 
                    baseQuery, watermarkColumn, lastWatermark.toString());
        }
        return baseQuery;
    }

    @Override
    public Map<String, Object> read() {
        if (rowIterator == null || !rowIterator.hasNext()) {
            return null; // Menghentikan langkah batch Spring
        }

        FieldValueList row = rowIterator.next();
        Map<String, Object> mappedRow = convertRowToMap(row);
        trackMaxWatermark(mappedRow);
        return mappedRow;
    }

    private Map<String, Object> convertRowToMap(FieldValueList row) {
        Map<String, Object> map = new LinkedHashMap<>();
        for (Field field : tableSchema.getFields()) {
            String colName = field.getName();
            FieldValue val = row.get(colName);
            if (val.isNull()) {
                map.put(colName, null);
                continue;
            }
            
            Object parsedVal = switch (field.getType().getStandardType()) {
                case TIMESTAMP -> val.getTimestampInstant();
                case INT64 -> val.getLongValue();
                case FLOAT64 -> val.getDoubleValue();
                case BOOL -> val.getBooleanValue();
                default -> val.getStringValue();
            };
            map.put(colName, parsedVal);
        }
        return map;
    }

    private void trackMaxWatermark(Map<String, Object> row) {
        if (watermarkColumn != null && row.get(watermarkColumn) instanceof Instant instantVal) {
            if (maxWatermarkSeen == null || instantVal.isAfter(maxWatermarkSeen)) {
                maxWatermarkSeen = instantVal;
            }
        }
    }

    @Override
    public void update(ExecutionContext executionContext) {
        // Menyimpan nilai watermark maksimum ke context untuk pemulihan kegagalan
        if (maxWatermarkSeen != null) {
            executionContext.putString("MAX_WATERMARK_SEEN", maxWatermarkSeen.toString());
        }
    }

    @Override
    public void close() {}
}

3. Postgres Upsert Writer Berkinerja Tinggi

Dalam database operasional, mengeksekusi kueri INSERT atau UPDATE satu per satu sangatlah lambat. Untuk mengoptimalkannya, kita menggabungkan baris data ke dalam batch transaksi (chunk-oriented processing misalnya 1000 item per transaksi).

Dalam Merge (Upsert) Mode, mesin data akan secara dinamis menyusun klausa SQL PostgreSQL: ON CONFLICT (primary_keys) DO UPDATE SET. Metode ini memastikan bahwa entri analitik baru dimasukkan langsung ke database, sedangkan data yang sudah ada (seperti pembaruan skor pengguna) diperbarui di tempat (in-place) tanpa memicu kesalahan pelanggaran batasan unik kunci.

import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;
import org.springframework.jdbc.core.JdbcTemplate;
import java.util.*;
import java.util.stream.Collectors;

public class PostgresItemWriter implements ItemWriter<Map<String, Object>> {

    private final JdbcTemplate jdbcTemplate;
    private final String tableName;
    private final List<String> primaryKeys;

    public PostgresItemWriter(JdbcTemplate jdbcTemplate, String tableName) {
        this.jdbcTemplate = jdbcTemplate;
        this.tableName = tableName;
        this.primaryKeys = List.of("id"); // Konfigurasi kunci primer target
    }

    @Override
    public void write(Chunk<? extends Map<String, Object>> chunk) {
        List<? extends Map<String, Object>> items = chunk.getItems();
        if (items.isEmpty()) return;

        List<String> columns = new ArrayList<>(items.getFirst().keySet());
        String upsertSql = buildUpsertSql(columns);

        List<Object[]> batchArgs = new ArrayList<>();
        for (Map<String, Object> row : items) {
            Object[] values = new Object[columns.size()];
            for (int i = 0; i < columns.size(); i++) {
                values[i] = row.get(columns.get(i));
            }
            batchArgs.add(values);
        }

        jdbcTemplate.batchUpdate(upsertSql, batchArgs);
    }

    private String buildUpsertSql(List<String> columns) {
        String colList = String.join(", ", columns);
        String placeholders = String.join(", ", Collections.nCopies(columns.size(), "?"));
        
        String conflictTarget = String.join(", ", primaryKeys);
        
        // Membangun klausa UPDATE, abaikan kolom kunci primer dari perubahan
        String updateSet = columns.stream()
                .filter(col -> !primaryKeys.contains(col))
                .map(col -> String.format("%s = EXCLUDED.%s", col, col))
                .collect(Collectors.joining(", "));

        return String.format(
                "INSERT INTO %s (%s) VALUES (%s) ON CONFLICT (%s) DO UPDATE SET %s",
                tableName, colList, placeholders, conflictTarget, updateSet
        );
    }
}

Keputusan Arsitektur: Pemilihan Infrastruktur secara Pragmatis

Meskipun arsitektur data modern menawarkan platform ETL khusus (seperti GCP Dataflow/Apache Beam atau SaaS Reverse ETL), kami memilih pendekatan rekayasa yang pragmatis dan hemat sumber daya:

  1. Memanfaatkan Aset yang Sudah Ada: Daripada membuat VM baru, mengonfigurasi kebijakan IAM GCP yang rumit, atau menyiapkan pipa rilis CI/CD terpisah, kami memilih untuk memperluas fungsionalitas layanan etl-service berbasis Java/Spring Boot yang sudah berjalan stabil di AstraPay.
  2. Meminimalkan Beban Operasional (Operational Overhead): Menambahkan tugas (job/step) baru ke dalam layanan yang sudah terwadahi (containerized) memungkinkan Reverse ETL berjalan menggunakan kapasitas server saat ini. Hal ini mencegah pembengkakan VM baru dan mempermudah pemeliharaan jangka panjang oleh tim platform.
  3. Kedaulatan Data & Kepatuhan Regulasi: Membangun solusi internal memastikan bahwa data finansial sensitif pengguna tetap berada di dalam jaringan VPC privat perusahaan, mematuhi standar kepatuhan regulasi tekfin lokal, serta mengeliminasi biaya lisensi platform pihak ketiga.

Pemulihan Transaksi & Ketahanan Kegagalan

Spring Batch menyediakan mekanisme manajemen transaksi yang tangguh:

  • Chunk Commits: Dengan membagi baris data menjadi chunk terpisah, apabila terjadi putusnya koneksi jaringan ketika menulis ke PostgreSQL, Spring Batch hanya akan membatalkan (rollback) chunk yang sedang berjalan. Chunk sebelumnya tetap tersimpan dengan aman di database.
  • Metadata Recovery: Dengan menuliskan stempel waktu maxWatermarkSeen ke dalam ExecutionContext Spring Batch saat langkah berjalan, status pekerjaan akan selalu disimpan. Jika terjadi mati listrik atau kegagalan server total, proses berikutnya dapat membaca status dari log context dan melanjutkan sinkronisasi dari posisi terakhir.

Dampak Bisnis & Hasil Kerja

  • Penghematan GCP: Migrasi dari metode Full Load harian menjadi penyaringan delta berbasis watermark berhasil meminimalkan volume pemindaian data di BigQuery secara signifikan.
  • Keamanan Database Operasional: Mengelompokkan transaksi penulisan batch meminimalkan risiko deadlock pada tabel relasional utama PostgreSQL dan menjaga rata-rata latensi transaksi di bawah 320ms per chunk.
  • Kecepatan Tindak Lanjut CRM: Segmentasi pengguna hasil kalkulasi machine learning di BigQuery kini dapat dialirkan ke sistem operasional dalam hitungan jam, mempercepat tim pemasaran untuk memicu kampanye personal kepada pengguna.