Incremental processing: a practical guide for developers

Introduction

Most data pipelines process everything from scratch on every run—wasteful, slow, and expensive. Incremental processing: a practical guide for developers takes a different approach. Instead of reprocessing months of historical data when you only need yesterday's changes, we selectively process just what's new or modified. For teams running Spark on tight schedules and budgets, mastering incremental patterns means faster jobs, lower cloud costs, and more reliable pipelines.

Over the next few sections, we'll walk through four core techniques: watermarks to track progress, change data capture to detect what actually changed, partition overwrites to minimize writes, and backfill strategies for handling corrections and gaps.

Watermarks — Track the last processed timestamp

A watermark is simply the highest timestamp you've successfully processed so far. On the next job run, you start from that watermark instead of the beginning of time. It's the simplest and most common way to implement incremental processing.

The pattern looks like this: store a metadata table (or file) that records the last processed timestamp. Before running your transformation, read that timestamp. Then filter your source data for anything newer. After processing completes, update the watermark.

Here's a practical example. Suppose we have a user events table in S3, partitioned by date. We want to aggregate events daily but only process new events since our last run.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, max as spark_max
from datetime import datetime
import json

spark = SparkSession.builder.appName("IncrementalEvents").getOrCreate()

# Read the last watermark
def read_watermark(watermark_path):
    try:
        with open(watermark_path, 'r') as f:
            data = json.load(f)
            return data.get('last_processed_timestamp')
    except FileNotFoundError:
        return None

# Update the watermark
def write_watermark(watermark_path, timestamp):
    with open(watermark_path, 'w') as f:
        json.dump({'last_processed_timestamp': timestamp}, f)

watermark_path = '/tmp/events_watermark.json'
last_timestamp = read_watermark(watermark_path)

# Read source events
events = spark.read.parquet('s3://data-bucket/events/')

# Filter for only new events
if last_timestamp:
    events_to_process = events.filter(col('event_timestamp') > last_timestamp)
else:
    events_to_process = events

# Process and aggregate
daily_agg = events_to_process.groupBy('event_date').agg(
    spark_max('user_id').alias('max_user_id'),
    spark_max('event_timestamp').alias('max_timestamp')
)

# Get the new watermark
new_watermark = events_to_process.agg(spark_max('event_timestamp')).collect()[0][0]

# Write results
daily_agg.write.mode('append').parquet('s3://data-bucket/daily_agg/')

# Update watermark only after successful write
if new_watermark:
    write_watermark(watermark_path, str(new_watermark))

Key points: store your watermark somewhere your job can reliably read and write it (databases work better than files for distributed environments). Always update the watermark after confirming your writes succeeded. If your job fails halfway through, the watermark stays at the old value and you'll reprocess on the next run—exactly what you want.

Change data capture — Read changes from source databases

Watermarks work well for append-only logs, but what if your source is a transactional database where rows get updated or deleted? That's where change data capture (CDC) comes in. CDC systems track insert, update, and delete operations at the source database level, letting you consume only those changes downstream.

Common CDC sources include Postgres logical decoding, MySQL binlog, and managed services like AWS DMS. Spark can read CDC streams via Kafka, files, or APIs. The tricky part is handling all three operation types correctly.

Let's imagine we sync a customer table from Postgres to a data lake. The Postgres connector outputs changes as a Kafka topic where each message includes the operation type.

from pyspark.sql.functions import col, when, row_number
from pyspark.sql.window import Window

# Read CDC stream from Kafka
cdc_messages = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "postgres.public.customers") \
    .load()

# Parse the CDC payload (assuming Debezium format)
from pyspark.sql.types import StructType, StructField, StringType, LongType

schema = StructType([
    StructField("op", StringType()),  # 'c'=create, 'u'=update, 'd'=delete
    StructField("before", StructType([
        StructField("id", LongType()),
        StructField("name", StringType())
    ])),
    StructField("after", StructType([
        StructField("id", LongType()),
        StructField("name", StringType())
    ]))
])

from pyspark.sql.functions import from_json

cdc_parsed = cdc_messages.select(
    from_json(col("value").cast("string"), schema).alias("change")
).select("change.*")

# Convert to upsert format: extract the final state and operation
changes = cdc_parsed.select(
    when(col("op") == "d", col("before.id")).otherwise(col("after.id")).alias("customer_id"),
    when(col("op") == "d", None).otherwise(col("after.name")).alias("name"),
    col("op").alias("operation")
)

# Write to a change table
changes.writeStream \
    .format("parquet") \
    .option("path", "s3://data-bucket/customer_changes/") \
    .option("checkpointLocation", "s3://data-bucket/checkpoint/") \
    .start()

When you read this change table downstream, you need to apply the deletes and updates properly. If you're building a slowly-changing dimension or maintaining a current state table, use MERGE (if your target supports it) or build a deduplication logic that keeps the latest version of each record.

Partition overwrite — Rewrite only affected partitions

Suppose your daily pipeline discovers that yesterday's data had errors. Do you rewrite the entire dataset or just yesterday's partition? Smart teams use overwrite mode with partition predicates to rewrite only what changed.

Spark's dynamic partition overwrite (available since 2.4.0 with spark.sql.sources.partitionOverwriteMode=dynamic) automatically detects which partitions your new data touches and overwrites only those. This is much faster than full overwrites.

from pyspark.sql.functions import col, current_date

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

# Read today's raw events
raw_events = spark.read.parquet('s3://data-bucket/raw_events/')

# Our target table is partitioned by event_date
processed = raw_events.filter(
    col('event_date') >= current_date() - 2  # reprocess last 2 days
).withColumn('processing_date', current_date())

# Write with dynamic partition overwrite
# Only partitions touched by the new data will be rewritten
processed.write \
    .mode('overwrite') \
    .partitionBy('event_date') \
    .parquet('s3://data-bucket/processed_events/')

Without dynamic mode, Spark would delete the entire table and write it fresh—expensive and slow. With it, only the partitions in your new data are touched. This is crucial for large tables where a full rewrite takes hours.

Backfills — Design jobs to reprocess date ranges

Incremental pipelines break when you need to reprocess a date range. Maybe you discovered a bug in your logic that affected the past month, or a data quality issue requires correction. Good backfill design means you can run the same job over any date range without breaking your production pipeline.

The trick is parameterizing your date range. Instead of hardcoding "process yesterday," allow jobs to accept a start and end date. Your watermark-based job becomes a special case: start from last watermark, end at today.

from pyspark.sql.functions import col
from datetime import datetime, timedelta

def run_daily_process(spark, start_date, end_date):
    """
    Process events between start_date and end_date (inclusive).
    start_date and end_date should be datetime objects.
    """
    
    # Read all events
    events = spark.read.parquet('s3://data-bucket/events/')
    
    # Filter for the date range
    filtered = events.filter(
        (col('event_date') >= start_date) & 
        (col('event_date') <= end_date)
    )
    
    # Apply transformations
    daily_agg = filtered.groupBy('event_date').agg(
        count('*').alias('event_count'),
        max('user_id').alias('max_user_id')
    )
    
    # Write (overwrite mode for backfills)
    daily_agg.write.mode('overwrite').partitionBy('event_date') \
        .parquet('s3://data-bucket/daily_agg/')

# For production: process yesterday
yesterday = datetime.now() - timedelta(days=1)
run_daily_process(spark, yesterday, yesterday)

# For backfill: process a whole month
start = datetime(2024, 1, 1)
end = datetime(2024, 1, 31)
run_daily_process(spark, start, end)

Document the date range parameters in your job. Make them easy to override from your scheduler (Airflow, Cron, etc.). Backfills should not update your watermark—only your incremental production run should do that.

Common mistakes

• Updating the watermark before writes complete. If your write fails after updating the watermark, you've lost data. Always update after confirming success.
• Assuming your source data is perfectly ordered. Late-arriving data can sneak in after your watermark. Either build a grace period (process the last N days on each run) or use CDC for guaranteed ordering.
• Forgetting to handle gaps after job failures. If your job crashes for 3 days, your watermark is stale. Plan for backfill logic to catch up cleanly.
• Not testing backfill scenarios. Your incremental logic might work fine for small daily runs but break when processing a month's data at once. Test both.
• Using files for watermarks in distributed setups. Consistency is hard. Use a database or object store with strong consistency (Postgres, DynamoDB) instead.

Key takeaways

• Watermarks track the highest timestamp processed; they're simple and work for append-only sources.
• Change data capture handles updates and deletes by streaming row-level changes from your source database.
• Dynamic partition overwrite rewrites only the partitions your new data touches, dramatically faster than full rewrites.
• Parameterize your date range so the same job handles daily incremental runs and historical backfills.
• Always update your watermark after confirming writes succeed, never before.
• Plan for late-arriving data by reprocessing a small window (last 2–3 days) on each run to catch stragglers.