Building an Incremental Watermark-Driven Reverse ETL with Spring Batch

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

Executive Summary

In modern data architectures, data pipelines are no longer a one-way street leading to a data warehouse. Processed analytical insights generated in a data warehouse often need to be synchronized back into downstream operational databases to support transactional requirements, service personalization, and real-time operational needs. This process is known as Reverse ETL. As a case study, I am taking the architecture implemented at AstraPay, where processed analytical insights from Google Cloud BigQuery are synchronized back into Cloud SQL PostgreSQL operational databases to improve data processing efficiency.

Re-syncing data via a full-table reload (Full Load) is highly inefficient. It creates unnecessary write bottlenecks on the operational transactional database, consumes significant CPU/network resources, and leads to an increase in query scan charges on data warehouse.

To resolve these bottlenecks, I designed and engineered a standalone Reverse ETL engine using Spring Boot and Spring Batch—an enterprise-grade Java framework designed specifically for processing large volumes of records with out-of-the-box features like chunk-oriented processing, transaction management, and fault tolerance.

In simple terms, Spring Batch divides a large task (Job) into several phases (Steps). In a chunk-based Step, records are read one by one from the source, accumulated in an in-memory buffer up to a certain size limit (e.g., 1000 items), and then written as a bulk upsert to the target database in a single transaction. This processing model can be visualized as follows:

graph TD
    subgraph Job [Spring Batch Job]
        subgraph Step [Step: Chunk-Oriented Processing]
            Reader[ItemReader <br> Reads records one by one from BigQuery] -->|1. Read record by record| Buffer(Chunk Buffer)
            Buffer -->|2. When chunk size is reached e.g. 1000| Writer[ItemWriter <br> Writes bulk upsert to PostgreSQL in a single transaction]
        end
    end

The core architectural pattern uses a Watermark Column Scan to perform delta detection, enabling the pipeline to operate in three modes: Full Load, Incremental, and Merge/Upsert.

---

Ingestion Flow & Architecture

The synchronization flow is structured around a central metadata engine that queries the target database for the most recent state before pulling delta records from the 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. Watermark Extraction: Before streaming data, the batch job inspects the target PostgreSQL operational table to fetch the maximum sync timestamp.
2. Lazy Dynamic Ingestion: Using this resolved watermark value, a custom Spring Batch ItemReader dynamically generates and executes a filtered standard SQL query on BigQuery.
3. Resilient Chunk-based Writing: The retrieved rows are streamed in transactional chunks and merged into the destination Postgres database using high-performance SQL upsert commands, tracking the maximum processed timestamp along the way.

---

Key Architectural Implementations

Note: The sequence diagram above and the code snippets below illustrate the Incremental/Merge Mode (watermark-driven delta sync), which represents the core architectural complexity of the engine. The implementations have been simplified for readability and sanitized of proprietary corporate logic.

1. Lazy Watermark Resolution

A common pitfall in Spring Batch configuration is executing metadata queries during the Spring Application Context initialization (bean definition). Doing so can lead to initialization errors if target databases are temporarily unreachable or if the watermark changes between application startup and job execution.

To solve this, I designed a Lazy Watermark Supplier. The query to fetch the maximum timestamp (SELECT MAX(watermark_col)) is wrapped inside a lazy Java Supplier<Instant>. The database is only queried when the batch step actually starts running, not when Spring sets up the beans:

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) {
        // Defaulting to "updated_at" for this Incremental/Merge mode demonstration
        String watermarkColumn = "updated_at";
        // The supplier resolves the watermark lazily when the job executes
        Supplier<Instant> watermarkSupplier = () -> resolveLastWatermark(table, watermarkColumn);
        
        return new BigQueryItemReader(
                bigQueryClient, 
                dataset, 
                table, 
                watermarkColumn, 
                watermarkSupplier
        );
    }

    private Instant resolveLastWatermark(String tableName, String watermarkColumn) {
        // SQL identifier sanitization is critical to prevent 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) {
            // Fallback to null (triggers a full load) if table is empty or error occurs
            return null;
        }
    }

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

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

2. Streaming BigQuery Item Reader

Querying millions of rows from BigQuery and loading them into JVM memory will instantly trigger OutOfMemoryError crashes.

The custom BigQueryItemReader implements Spring’s ItemReader and ItemStream interfaces. It queries BigQuery using standard SQL pagination or Google’s native result set streaming iterator (TableResult.iterateAll()). The reader translates each unstructured BigQuery row (FieldValueList) into a standard Java Map<String, Object> and records the maximum watermark timestamp encountered in the current chunk.

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) {
        // Lazy resolution of target watermark prior to executing query
        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("BigQuery query execution interrupted", 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; // Signals Spring Batch to complete the step
        }

        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) {
        // Persist max watermark seen to execution context for resiliency
        if (maxWatermarkSeen != null) {
            executionContext.putString("MAX_WATERMARK_SEEN", maxWatermarkSeen.toString());
        }
    }

    @Override
    public void close() {}
}

3. High-Performance SQL Upsert Writer

In transactional databases, executing millions of individual INSERT or UPDATE queries is highly inefficient. Instead, we bundle rows in standard transactional chunks (e.g., 1000 items) and perform batch operations.

In Merge (Upsert) Mode, the engine generates a PostgreSQL-specific ON CONFLICT (primary_keys) DO UPDATE SET statement. This ensures new analytical entries are inserted immediately, while existing records (like updated merchant scores) are overwritten in-place without causing unique constraint violations.

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"); // Configured primary key 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);
        
        // Construct the UPDATE clause, excluding primary keys from modification
        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
        );
    }
}

---

Architectural Decision: Pragmatic Infrastructure Selection

While modern data architectures offer specialized platforms (such as GCP Dataflow/Apache Beam or dedicated SaaS tools), we opted for a highly pragmatic, resource-efficient engineering approach:

1. Leveraging Existing Assets: Rather than provisioning new virtual machines, configuring complex GCP IAM policies, or setting up separate CI/CD pipelines, we chose to extend AstraPay’s existing, stable Java/Spring Boot etl-service.
2. Minimizing Operational Overhead: Adding a new job and step to an already containerized service allowed us to execute the Reverse ETL within our current resource footprint. This avoided VM sprawl and simplified long-term maintenance for the platform team.
3. Data Governance & Compliance: Keeping the integration in-house ensured that customer financial metadata remained entirely inside our private VPC, complying with local fintech regulations and avoiding the costs of third-party SaaS solutions.

---

Resilience, Transactions, & Fault Tolerance

Spring Batch provides robust transaction management out of the box:

• Chunk Commits: By dividing the stream into chunks (e.g., 1000 rows), if a network timeout occurs while writing to PostgreSQL, Spring Batch rolls back the current chunk only. The previous chunk remains safely committed.
• Metadata Recovery: By writing the current maxWatermarkSeen value into the Spring Batch ExecutionContext during the update() hook, the job state is saved. In the event of a total engine crash, the next run can restore the state from the execution context metadata and resume the synchronization step.

---

Business Impact & Results

• GCP Cost Reduction: Shifting from daily Full Load sweeps to dynamic watermark-driven incremental merges decreased daily query scan volumes on BigQuery significantly.
• Reduced Database Locking: Batching writes into transaction-bound upserts eliminated deadlocks on PostgreSQL transactional tables and kept write latency under 320ms per chunk.
• Operational Agility: Business intelligence and CRM segment definitions processed in the BigQuery warehouse are synchronized to backend systems within hours, allowing the product team to automate campaign triggers.