Checkpointing and State Management in Long-Running Data Pipelines

Introduction

It's 2 AM on a Tuesday. Your pipeline has been processing 48 hours of clickstream data — 800 million events, transforming and enriching each one against a product catalog and user profile service. It's been running for six hours. Then a database connection times out, a memory limit gets hit, or a spot instance gets reclaimed by the cloud provider. The job crashes. You stare at the logs, and the question hits you: do you start over from the beginning, or do you have some way to resume from where things left off?

If you don't have a checkpointing strategy, you restart from scratch. You burn another six hours of compute, you delay the downstream teams waiting on this data, and you explain to your manager why the morning dashboard is still showing yesterday's numbers. If you do have a checkpointing strategy, you restart from the last safe point, process the remaining work, and are done before your first cup of coffee.

This lesson is about building pipelines that are resilient by design. We'll cover how checkpointing works at a conceptual level, how to implement it in Python-based pipelines with realistic patterns, how Spark handles stateful streaming, and how to manage accumulated state over time without it becoming a liability. By the end, you'll be able to design and implement checkpoint-aware pipelines that can survive failure without losing work or producing incorrect results.

What you'll learn:

• The difference between checkpointing, state management, and idempotency — and why you need all three
• How to implement file-based and database-backed checkpointing for batch pipelines in Python
• How Apache Spark manages checkpoints and streaming state internally, and what you need to configure
• How to handle state expiration to prevent unbounded growth in long-running jobs
• How to debug corrupted or stale checkpoints that cause pipelines to produce wrong results

---

Prerequisites

You should be comfortable with:

• Python at an intermediate level (classes, context managers, exception handling)
• Basic SQL and familiarity with a database (Postgres, SQLite, or similar)
• The concept of ETL/ELT pipelines and data partitioning
• Passing familiarity with Apache Spark (helpful for the Spark sections, but not required)

---

The Core Problem: Why Pipelines Fail Partway Through

Before writing any code, let's be precise about what we're actually solving. Long-running pipelines fail for reasons that are largely outside your control:

• Transient infrastructure failures: spot instance preemption, network blips, cloud provider hiccups
• Resource exhaustion: OOM kills, disk-full errors, connection pool saturation
• Upstream data problems: a malformed record deep in the dataset that causes a parsing exception
• Time limits: cloud function timeouts, cluster autoscaling events, scheduled job preemptions

What all of these have in common is that they're unexpected interruptions to a process that was otherwise making progress. The naive response — "just retry the whole thing" — is expensive at best and incorrect at worst.

Consider a pipeline that:

1. Reads 10 million customer transactions
2. Enriches each with the current exchange rate for that currency
3. Writes the results to a data warehouse

If step 3 fails at record 7.8 million, a blind retry will re-enrich and re-write records 1 through 7.8 million. If your write operation is not idempotent (meaning: running it twice produces duplicate rows), you now have bad data. Checkpointing and idempotency are a package deal.

Key insight: Checkpointing tells you where you left off. Idempotency ensures that re-processing work you already completed doesn't break anything. You need both.

---

Checkpointing Concepts: What to Save and When

A checkpoint is a durable record of your pipeline's progress. "Durable" is the operative word — saving state to an in-process dictionary doesn't help when the process dies.

There are three things worth checkpointing, depending on your pipeline:

1. Offset or cursor position The simplest form: a record of what you've processed. "I've processed all transactions with transaction_id <= 7,823,441" or "I've processed all files in the S3 prefix s3://data-lake/transactions/2024/01/ up through batch_047.parquet."

2. Aggregated state For pipelines that compute running totals, rolling windows, or accumulated counts — you need to store the intermediate result, not just the position. "As of batch 47, the running total for customer usr_00293 is 4,230.88 USD."

3. Processing metadata Information about the work itself: which records failed and why, how long each batch took, what schema version was seen. This powers observability and debugging.

Checkpoint Granularity

How often you checkpoint is a tradeoff between overhead and recovery cost. If you process 10 million records in 500-record batches and checkpoint after every batch, a failure costs you at most one batch of reprocessing. If you checkpoint every 10,000 batches, a failure could cost you up to 5 million records of reprocessing.

For most batch pipelines, checkpointing every 1,000–10,000 records (or every 30–120 seconds of wall time) is a reasonable starting point. Adjust based on the cost of re-processing versus the overhead of checkpointing.

---

Implementing File-Based Checkpointing in Python

Let's build a concrete, production-quality checkpoint system for a batch pipeline. We'll process a large CSV export of e-commerce order events, enrich each event with product category data from a lookup table, and write the results to a Parquet file.

The Checkpoint Store

We need a checkpoint store that:

• Persists state to disk (survives process restarts)
• Supports atomic updates (no partial writes)
• Is human-readable (you want to be able to inspect it when debugging)

JSON files with atomic rename-writes are a good fit for single-machine pipelines:

import json
import os
import tempfile
from pathlib import Path
from datetime import datetime, timezone
from typing import Any, Optional


class FileCheckpointStore:
    """
    A durable checkpoint store backed by a JSON file.
    Uses atomic write-rename to prevent partial/corrupt checkpoint files.
    """

    def __init__(self, checkpoint_path: str):
        self.checkpoint_path = Path(checkpoint_path)
        self.checkpoint_path.parent.mkdir(parents=True, exist_ok=True)

    def load(self) -> Optional[dict]:
        """Load the current checkpoint. Returns None if no checkpoint exists."""
        if not self.checkpoint_path.exists():
            return None
        try:
            with open(self.checkpoint_path, "r") as f:
                return json.load(f)
        except (json.JSONDecodeError, IOError) as e:
            raise CheckpointCorruptionError(
                f"Checkpoint at {self.checkpoint_path} is unreadable: {e}"
            ) from e

    def save(self, state: dict) -> None:
        """
        Atomically save checkpoint state.
        Writes to a temp file first, then renames — prevents partial writes
        from corrupting the checkpoint if the process dies mid-write.
        """
        state["_checkpoint_written_at"] = datetime.now(timezone.utc).isoformat()
        
        # Write to a temp file in the same directory (same filesystem = atomic rename)
        dir_path = self.checkpoint_path.parent
        with tempfile.NamedTemporaryFile(
            mode="w",
            dir=dir_path,
            suffix=".tmp",
            delete=False
        ) as tmp_file:
            json.dump(state, tmp_file, indent=2)
            tmp_path = tmp_file.name

        # Atomic rename
        os.replace(tmp_path, self.checkpoint_path)

    def clear(self) -> None:
        """Remove the checkpoint (call this when a pipeline completes successfully)."""
        if self.checkpoint_path.exists():
            self.checkpoint_path.unlink()


class CheckpointCorruptionError(Exception):
    pass

The os.replace() call is the critical piece here. On POSIX systems, rename is atomic — the destination file will either be the old version or the new version, never a partial write. On Windows, os.replace() achieves the same effect. This means you'll never read a half-written checkpoint.

The Pipeline with Checkpointing

Now let's build the actual pipeline. We're processing a file of order events (imagine 50 million rows, split across daily files) and enriching each order with its product category.

import csv
import time
import logging
from pathlib import Path
from typing import Iterator
import pyarrow as pa
import pyarrow.parquet as pq

logger = logging.getLogger(__name__)


def load_product_categories(lookup_path: str) -> dict:
    """Load product_id -> category mapping into memory."""
    categories = {}
    with open(lookup_path, "r") as f:
        reader = csv.DictReader(f)
        for row in reader:
            categories[row["product_id"]] = row["category"]
    return categories


def iter_order_events(
    source_path: str,
    start_after_row: int = 0
) -> Iterator[tuple[int, dict]]:
    """
    Yields (row_number, row_dict) tuples from the order event CSV.
    If start_after_row > 0, skips the first N rows (checkpoint resume).
    """
    with open(source_path, "r") as f:
        reader = csv.DictReader(f)
        for row_num, row in enumerate(reader, start=1):
            if row_num <= start_after_row:
                continue
            yield row_num, row


def enrich_order_event(row: dict, categories: dict) -> dict:
    """Add product category to an order event row."""
    product_id = row.get("product_id", "")
    return {
        **row,
        "product_category": categories.get(product_id, "unknown"),
        "enriched_at": datetime.now(timezone.utc).isoformat(),
    }


def process_order_events(
    source_path: str,
    output_path: str,
    lookup_path: str,
    checkpoint_path: str,
    batch_size: int = 5_000,
    checkpoint_interval: int = 10,  # checkpoint every N batches
):
    checkpoint_store = FileCheckpointStore(checkpoint_path)
    categories = load_product_categories(lookup_path)

    # Resume from checkpoint if one exists
    checkpoint = checkpoint_store.load()
    if checkpoint:
        last_processed_row = checkpoint["last_processed_row"]
        rows_processed = checkpoint["rows_processed"]
        logger.info(
            f"Resuming from checkpoint: row {last_processed_row}, "
            f"{rows_processed} rows processed so far"
        )
    else:
        last_processed_row = 0
        rows_processed = 0
        logger.info("No checkpoint found, starting from the beginning")

    batch = []
    last_checkpoint_batch = 0
    output_file = Path(output_path)

    # If resuming, open in append mode; otherwise start fresh
    write_mode = "append" if checkpoint else "write"
    pq_writer = None

    try:
        for row_num, row in iter_order_events(source_path, start_after_row=last_processed_row):
            enriched = enrich_order_event(row, categories)
            batch.append(enriched)

            if len(batch) >= batch_size:
                # Write the batch
                table = pa.Table.from_pylist(batch)
                if pq_writer is None:
                    pq_writer = pq.ParquetWriter(output_path, table.schema)
                pq_writer.write_table(table)

                rows_processed += len(batch)
                last_processed_row = row_num
                last_checkpoint_batch += 1
                batch = []

                # Checkpoint periodically
                if last_checkpoint_batch >= checkpoint_interval:
                    checkpoint_store.save({
                        "last_processed_row": last_processed_row,
                        "rows_processed": rows_processed,
                        "source_path": source_path,
                    })
                    logger.info(
                        f"Checkpoint saved at row {last_processed_row} "
                        f"({rows_processed} total rows processed)"
                    )
                    last_checkpoint_batch = 0

        # Write any remaining records in the final partial batch
        if batch:
            table = pa.Table.from_pylist(batch)
            if pq_writer is None:
                pq_writer = pq.ParquetWriter(output_path, table.schema)
            pq_writer.write_table(table)
            rows_processed += len(batch)

        logger.info(f"Pipeline complete. Total rows processed: {rows_processed}")

        # Clear the checkpoint on successful completion
        checkpoint_store.clear()

    except Exception as e:
        logger.error(f"Pipeline failed at row {last_processed_row}: {e}")
        # Save a final checkpoint so we can resume
        checkpoint_store.save({
            "last_processed_row": last_processed_row,
            "rows_processed": rows_processed,
            "source_path": source_path,
            "failed_at": datetime.now(timezone.utc).isoformat(),
            "error": str(e),
        })
        raise
    finally:
        if pq_writer:
            pq_writer.close()

There are a few important design decisions here worth calling out:

• The checkpoint is saved after the write, not before. If we saved the checkpoint position before writing the batch and then the write failed, we'd skip those records on resume. Write first, then checkpoint.
• The output uses append mode on resume. This only works correctly if the writer is idempotent at the batch level — Parquet row groups work fine here because we're appending, not overwriting.
• The checkpoint is cleared on success. This prevents a stale checkpoint from confusing the next run.

Warning: Row-number-based checkpointing only works if the source file is immutable (not being modified during processing). If the source can change, use a stable row identifier instead — an order ID, a database primary key, or a content hash.

---

Database-Backed Checkpointing for Distributed Pipelines

File-based checkpointing works well for single-machine pipelines. For distributed pipelines where multiple workers are processing different partitions, you need a shared, concurrent-safe checkpoint store.

Postgres is a solid choice. Here's a checkpoint store backed by a simple Postgres table:

-- Run this once to set up the checkpoint table
CREATE TABLE IF NOT EXISTS pipeline_checkpoints (
    pipeline_id     TEXT NOT NULL,
    partition_key   TEXT NOT NULL,
    last_offset     BIGINT NOT NULL,
    records_processed BIGINT NOT NULL DEFAULT 0,
    state_json      JSONB,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (pipeline_id, partition_key)
);

CREATE INDEX IF NOT EXISTS idx_checkpoints_pipeline 
    ON pipeline_checkpoints (pipeline_id, updated_at DESC);

import psycopg2
import psycopg2.extras
from contextlib import contextmanager
from typing import Optional
import json


class PostgresCheckpointStore:
    """
    Checkpoint store backed by Postgres.
    Supports concurrent workers via partition-level checkpointing.
    """

    def __init__(self, dsn: str, pipeline_id: str):
        self.dsn = dsn
        self.pipeline_id = pipeline_id

    @contextmanager
    def _connection(self):
        conn = psycopg2.connect(self.dsn)
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

    def load(self, partition_key: str) -> Optional[dict]:
        """Load checkpoint for a specific partition."""
        query = """
            SELECT last_offset, records_processed, state_json
            FROM pipeline_checkpoints
            WHERE pipeline_id = %s AND partition_key = %s
        """
        with self._connection() as conn:
            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
                cur.execute(query, (self.pipeline_id, partition_key))
                row = cur.fetchone()
                if row is None:
                    return None
                return {
                    "last_offset": row["last_offset"],
                    "records_processed": row["records_processed"],
                    "state": row["state_json"] or {},
                }

    def save(self, partition_key: str, last_offset: int, 
             records_processed: int, state: dict = None) -> None:
        """
        Upsert checkpoint for a partition.
        Uses INSERT ... ON CONFLICT to handle concurrent workers safely.
        """
        query = """
            INSERT INTO pipeline_checkpoints 
                (pipeline_id, partition_key, last_offset, records_processed, state_json, updated_at)
            VALUES (%s, %s, %s, %s, %s, NOW())
            ON CONFLICT (pipeline_id, partition_key) 
            DO UPDATE SET
                last_offset = EXCLUDED.last_offset,
                records_processed = EXCLUDED.records_processed,
                state_json = EXCLUDED.state_json,
                updated_at = NOW()
        """
        with self._connection() as conn:
            with conn.cursor() as cur:
                cur.execute(query, (
                    self.pipeline_id,
                    partition_key,
                    last_offset,
                    records_processed,
                    json.dumps(state) if state else None,
                ))

    def list_partitions(self) -> list[dict]:
        """List all partition checkpoints for this pipeline (useful for monitoring)."""
        query = """
            SELECT partition_key, last_offset, records_processed, updated_at
            FROM pipeline_checkpoints
            WHERE pipeline_id = %s
            ORDER BY partition_key
        """
        with self._connection() as conn:
            with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur:
                cur.execute(query, (self.pipeline_id,))
                return [dict(row) for row in cur.fetchall()]

    def clear(self, partition_key: str = None) -> None:
        """Clear checkpoint(s). Pass partition_key to clear one, or None to clear all."""
        if partition_key:
            query = "DELETE FROM pipeline_checkpoints WHERE pipeline_id = %s AND partition_key = %s"
            params = (self.pipeline_id, partition_key)
        else:
            query = "DELETE FROM pipeline_checkpoints WHERE pipeline_id = %s"
            params = (self.pipeline_id,)

        with self._connection() as conn:
            with conn.cursor() as cur:
                cur.execute(query, params)

With this store, each worker in your distributed pipeline checkpoints its own partition independently. Worker processing partition_key = "2024-01-15" doesn't interfere with the worker processing partition_key = "2024-01-16". If the partition-15 worker dies, only that partition needs to be reprocessed.

---

Stateful Streaming with Apache Spark

Batch pipeline checkpointing is conceptually simple — you're recording a position in a file or a database. Streaming pipelines are harder because the state is live and evolving. You're continuously computing aggregations over windows of time, and those windows can overlap, get updated by late-arriving data, and span machine failures.

Spark Structured Streaming handles this with two mechanisms: checkpointing (for fault recovery) and state store (for maintaining aggregation state between micro-batches).

Configuring Spark Checkpointing

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import *

spark = SparkSession.builder \
    .appName("order-event-aggregator") \
    .config("spark.sql.shuffle.partitions", "50") \
    .getOrCreate()

# Read from Kafka (a realistic streaming source)
raw_orders = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-broker:9092") \
    .option("subscribe", "order-events") \
    .option("startingOffsets", "latest") \
    .option("failOnDataLoss", "false") \
    .load()

order_schema = StructType([
    StructField("order_id", StringType()),
    StructField("customer_id", StringType()),
    StructField("amount_usd", DoubleType()),
    StructField("event_time", TimestampType()),
    StructField("status", StringType()),
])

orders = raw_orders.select(
    F.from_json(F.col("value").cast("string"), order_schema).alias("data"),
    F.col("timestamp").alias("kafka_timestamp")
).select("data.*", "kafka_timestamp")

# Compute 1-hour windowed revenue totals per customer
# withWatermark tells Spark how late data can arrive before it's dropped
windowed_revenue = orders \
    .withWatermark("event_time", "30 minutes") \
    .groupBy(
        F.col("customer_id"),
        F.window(F.col("event_time"), "1 hour", "15 minutes")  # 1-hour windows, sliding every 15 min
    ) \
    .agg(
        F.sum("amount_usd").alias("total_revenue_usd"),
        F.count("order_id").alias("order_count"),
        F.max("event_time").alias("last_event_time"),
    )

# Write results, with checkpointing enabled
query = windowed_revenue.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://data-lake/checkpoints/order-revenue-agg/") \
    .option("path", "s3://data-lake/output/customer-hourly-revenue/") \
    .trigger(processingTime="2 minutes") \
    .start()

query.awaitTermination()

The checkpointLocation option is doing a lot of work here. Spark uses it to store:

• Offsets: The Kafka offset position for each partition, so on restart Spark knows exactly where to continue reading
• Committed batch IDs: Which micro-batches have been fully written, enabling exactly-once semantics with compatible sinks
• State store snapshots: The current aggregation state (the windowed revenue totals), snapshotted periodically

Critical: The checkpoint location must be on a reliable, durable store — S3, ADLS, or HDFS. Local filesystem is fine for development but will not survive node failures in production.

Understanding Watermarks and State Expiration

One of the most common production problems with stateful streaming is unbounded state growth. If you're computing windowed aggregations and you never tell Spark to drop old windows, it will keep accumulating state indefinitely until the executor runs out of memory.

The withWatermark("event_time", "30 minutes") call solves this. It tells Spark: "Don't hold state for any window that ended more than 30 minutes before the current watermark." The watermark advances as new data arrives — it's the maximum event time seen minus the specified delay.

Let's trace through a concrete example:

• At 14:00, the most recent event seen has event_time = 13:58. Watermark = 13:28.
• Windows that ended before 13:28 are finalized and their state is dropped.
• At 14:30, the most recent event has event_time = 14:29. Watermark = 13:59.
• An event arrives late with event_time = 13:15. Since 13:15 < watermark (13:59), it's dropped.

This is the fundamental tension in streaming: you want to wait long enough to catch late data, but not so long that your state grows without bound. 30 minutes is a common starting point. For data with high network latency or significant clock skew, you may need to go up to several hours.

---

Managing State in Custom Python Pipelines

Not every pipeline uses Spark. Many production pipelines are Python scripts running on Airflow, Prefect, or Dagster. For these, you need to manage intermediate state yourself.

Here's a realistic pattern: you're processing customer events and computing a running risk score for each customer. The score depends on all historical behavior, so you can't compute it from a single batch — you need to carry forward the previous score and update it incrementally.

from dataclasses import dataclass, asdict
from typing import Optional
import json


@dataclass
class CustomerRiskState:
    customer_id: str
    risk_score: float
    transaction_count: int
    failed_payment_count: int
    last_event_time: str
    last_updated: str

    def update_from_event(self, event: dict) -> "CustomerRiskState":
        """
        Apply a single event to update the risk state.
        Returns a new state object (immutable update pattern).
        """
        new_tx_count = self.transaction_count + 1
        new_failed = self.failed_payment_count + (
            1 if event.get("status") == "payment_failed" else 0
        )
        
        # Simplified risk scoring: failure rate weighted by recency
        failure_rate = new_failed / new_tx_count if new_tx_count > 0 else 0
        # New events have more weight — exponential smoothing
        new_score = 0.7 * self.risk_score + 0.3 * (failure_rate * 100)

        return CustomerRiskState(
            customer_id=self.customer_id,
            risk_score=round(new_score, 4),
            transaction_count=new_tx_count,
            failed_payment_count=new_failed,
            last_event_time=event["event_time"],
            last_updated=datetime.now(timezone.utc).isoformat(),
        )


class CustomerRiskStateStore:
    """
    Manages customer risk state with periodic persistence.
    Backed by a Postgres table, but cached in-memory for performance.
    """

    def __init__(self, checkpoint_store: PostgresCheckpointStore):
        self.checkpoint_store = checkpoint_store
        self._cache: dict[str, CustomerRiskState] = {}
        self._dirty: set[str] = set()  # track which records need flushing

    def get(self, customer_id: str) -> Optional[CustomerRiskState]:
        if customer_id in self._cache:
            return self._cache[customer_id]

        # Load from persistent store
        data = self.checkpoint_store.load(partition_key=f"risk:{customer_id}")
        if data and data.get("state"):
            state = CustomerRiskState(**data["state"])
            self._cache[customer_id] = state
            return state

        return None

    def update(self, customer_id: str, event: dict) -> CustomerRiskState:
        current = self.get(customer_id)
        if current is None:
            current = CustomerRiskState(
                customer_id=customer_id,
                risk_score=0.0,
                transaction_count=0,
                failed_payment_count=0,
                last_event_time=event["event_time"],
                last_updated=datetime.now(timezone.utc).isoformat(),
            )
        
        new_state = current.update_from_event(event)
        self._cache[customer_id] = new_state
        self._dirty.add(customer_id)
        return new_state

    def flush(self, force: bool = False) -> int:
        """
        Persist dirty state to the checkpoint store.
        Returns the number of states flushed.
        """
        flushed = 0
        for customer_id in list(self._dirty):
            state = self._cache[customer_id]
            self.checkpoint_store.save(
                partition_key=f"risk:{customer_id}",
                last_offset=0,  # not offset-based
                records_processed=state.transaction_count,
                state=asdict(state),
            )
            self._dirty.discard(customer_id)
            flushed += 1
        return flushed

    def evict_cold_state(self, max_cache_size: int = 100_000) -> None:
        """
        If the in-memory cache grows too large, evict the least-recently-accessed
        states (flush dirty ones first). This prevents memory exhaustion on large
        customer sets.
        """
        if len(self._cache) <= max_cache_size:
            return
        
        # Flush everything dirty before evicting
        self.flush()
        
        # Evict half the cache (simple strategy — use an LRU in production)
        evict_count = len(self._cache) // 2
        to_evict = list(self._cache.keys())[:evict_count]
        for key in to_evict:
            del self._cache[key]
        
        logger.info(f"Evicted {evict_count} entries from risk state cache")

The write-through cache pattern here is important for performance. If you're processing 10 million events across 2 million customers, hitting the database for every customer on every event would be catastrophically slow. Instead, you batch the writes and flush periodically.

Tip: Set a maximum cache size and implement eviction. It's easy to assume your customer set will fit in memory, and easy to be wrong at 3 AM on your biggest traffic day of the year.

---

Hands-On Exercise

Scenario: You're building a pipeline to process a daily export of 5 million retail transactions. The source is a CSV file partitioned by date (e.g., transactions_2024_01_15.csv). Each row contains: transaction_id, store_id, product_id, quantity, unit_price_usd, transaction_timestamp.

Your pipeline must:

1. Compute daily revenue totals per store
2. Compute a 7-day rolling revenue total per store (meaning you need to carry state across daily runs)
3. Write results to Parquet, partitioned by date
4. Be resumable if the daily run fails partway through

Exercise tasks:

Task 1: Implement a DailyRevenueState dataclass that holds store_id, daily_revenue, rolling_7d_revenue, a list of the last 7 days' daily totals, and last_updated. Implement an update(day_total: float, date: str) method that shifts the 7-day window correctly.

Task 2: Implement a TransactionPipelineCheckpoint class using the FileCheckpointStore from earlier that stores:

• The last processed transaction_id (for within-file resumption)
• The current store revenue states (so rolling totals survive across restarts)
• The processing date and source file path

Task 3: Write a process_daily_file(source_path, output_dir, checkpoint_path) function that:

• Loads the checkpoint if it exists
• Skips already-processed transactions (match on transaction_id, not row number — more robust)
• Checkpoints every 50,000 transactions
• On completion, writes the output Parquet and clears the checkpoint

Starter code:

import csv
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime, timezone
import pyarrow as pa
import pyarrow.parquet as pq

@dataclass
class DailyRevenueState:
    store_id: str
    daily_revenue: float = 0.0
    rolling_7d_revenue: float = 0.0
    last_7_days: list = field(default_factory=list)
    last_updated: str = ""

    def update(self, day_total: float, date: str) -> None:
        # YOUR IMPLEMENTATION HERE
        # Hint: append day_total to last_7_days, keep only the last 7,
        # recompute rolling_7d_revenue as the sum
        pass


def process_daily_file(
    source_path: str,
    output_dir: str,
    checkpoint_path: str,
    batch_size: int = 50_000,
) -> None:
    # YOUR IMPLEMENTATION HERE
    pass

Work through this exercise before looking up solutions. The core challenge is keeping the rolling 7-day window correct across restarts — make sure you're not double-counting a day's revenue if the pipeline fails mid-file and resumes.

---

Common Mistakes & Troubleshooting

Mistake 1: Checkpointing Before Writing

This is the most common correctness bug. If you save the checkpoint position before writing the batch to your output, and the write fails, your checkpoint claims you've processed those records but your output doesn't contain them. On resume, you skip them. Data is silently lost.

Fix: Always write output first. Only advance the checkpoint after a successful write.

Mistake 2: Not Making the Output Idempotent

Checkpointing handles the "where did I leave off" problem, but if your sink appends duplicate rows when re-processing, you'll have bad data even with correct checkpointing.

Fix: Use INSERT ... ON CONFLICT DO NOTHING in databases, or write to Parquet files keyed by partition and overwrite on reprocess. Delta Lake and Apache Iceberg provide transactional writes that handle this elegantly.

Mistake 3: Stale Checkpoints After Schema Changes

If you add a column to your output schema and your checkpoint stores state using the old schema, resuming from that checkpoint can cause type errors or missing fields.

Fix: Include a schema version number in your checkpoint state. On load, check the version. If it doesn't match, log a warning and either migrate the state or start fresh. A hard-to-debug production failure caused by a stale checkpoint from two months ago is not a fun morning.

CHECKPOINT_SCHEMA_VERSION = 2

def validate_checkpoint(checkpoint: dict) -> bool:
    version = checkpoint.get("schema_version", 1)
    if version != CHECKPOINT_SCHEMA_VERSION:
        logger.warning(
            f"Checkpoint schema version mismatch: expected {CHECKPOINT_SCHEMA_VERSION}, "
            f"got {version}. Starting fresh."
        )
        return False
    return True

Mistake 4: Using Row Numbers Instead of Stable Identifiers

Row-number-based checkpointing (last_processed_row = 7_823_441) is fragile. If anyone inserts or deletes rows in the source file — or if you're reading from a database with changing ordering — your row numbers shift and your checkpoint points to the wrong place.

Fix: Checkpoint against stable, immutable identifiers: primary keys, content hashes, Kafka offsets, or S3 file ETags.

Mistake 5: Spark Checkpoint Location Conflicts

In Spark Structured Streaming, if you change your query logic (add a column, change a window size) and restart without clearing the checkpoint, Spark may throw an error or silently use stale state.

Fix: When changing streaming logic significantly, use a new checkpoint location and perform a clean restart. Automate checkpoint location naming with a version suffix: s3://data-lake/checkpoints/order-revenue-agg/v3/.

Warning: Clearing a Spark checkpoint for a stateful aggregation job means you lose all accumulated window state. For long-running aggregation jobs, consider snapshotting state to a separate store before rotating the checkpoint.

Debugging Corrupted Checkpoints

Symptoms: pipeline starts but immediately fails with a deserialization error, produces wildly incorrect numbers, or skips large chunks of data.

Diagnostic steps:

1. Print the raw checkpoint content and visually inspect it
2. Check the _checkpoint_written_at timestamp — is it older than expected?
3. Compare last_processed_row or offset in the checkpoint against the actual source data
4. For Spark, examine the _spark_metadata directory in your checkpoint location — it contains JSON files with committed offset ranges

Recovery:

• For mild corruption: manually edit the JSON checkpoint to a known-good position
• For severe corruption: delete the checkpoint, identify the last successfully written output partition, and restart from there
• For Spark: delete the checkpoint directory and restart the query with startingOffsets = "earliest" if you can afford the recomputation, or restore from a state store backup

---

Summary & Next Steps

Let's consolidate what you've built:

Checkpointing is the practice of recording durable, resumable progress markers in long-running pipelines. The key decisions are: what to checkpoint (offset, state, metadata), how often (cost vs recovery granularity tradeoff), and where (file, database, distributed store).

Idempotency is the companion to checkpointing. A checkpoint tells you where to restart; idempotency ensures that restarting doesn't corrupt your output. These two properties together give you exactly-once processing semantics even in the face of arbitrary failures.

State management extends checkpointing to cover intermediate computation — running totals, rolling windows, per-entity scores — that need to survive restarts and scale to large datasets without exhausting memory.

The production patterns that matter most:

• Write-then-checkpoint ordering (never the reverse)
• Atomic writes for checkpoint files (rename-based)
• Schema versioning in checkpoint state
• Cache sizing with eviction for in-memory state stores
• Watermarks for bounded state in streaming pipelines

Next Steps

From here, your natural progressions are:

1. Delta Lake / Apache Iceberg: These table formats provide ACID transactions and built-in change tracking that make idempotent writes dramatically simpler. Understanding checkpointing at this level makes it much easier to understand what these formats are actually doing for you.

2. Kafka and offset management: Kafka's consumer offset management is a specialized checkpointing system. Now that you understand the concept deeply, studying Kafka consumer groups and commit semantics will make much more sense.

3. Stateful stream processing with Flink: Apache Flink's state backend and savepoint system is the most sophisticated stateful streaming system available. It handles everything we covered here — and more — at massive scale.

4. Workflow orchestration checkpointing: Airflow's XCom, Prefect's state system, and Dagster's asset materialization all implement forms of checkpoint-aware pipeline management. Understanding the underlying principles helps you use these tools more effectively and know when they're not enough.

The foundation you've built here — understanding why checkpoints exist and how they interact with output idempotency and state management — applies across all of these systems. The implementations differ, but the problems they're solving are identical.