Checkpoints¶

A checkpoint makes repeated batch reads resume where the last one stopped. On the first run, all the data is fetched, and the final pagination state is saved. Later runs resume from that state.

The checkpoint moves when the Spark task that read the table succeeds. A reading task that fails partway, is killed, or fails later in its own work (such as writing its output in the same task) leaves the checkpoint where it was, so the next run reads those records again instead of skipping them.

Some records can be read twice:

After a failed reading task, the next run starts from the previous checkpoint, so records the failed task had already delivered come again.
In cursor mode, the API's last page has no next cursor. The saved cursor is the one that fetched that page, so the next run reads it again, along with anything appended to it since. That's what keeps records added to the last page from being missed.

Not an end-to-end delivery guarantee

The checkpoint follows the reading task, not the whole query. If the query goes on past that task (after a shuffle, in a later stage, or in a sink's final commit) and fails there, the reading task has already succeeded and moved the checkpoint. Rerunning the query then starts after records that never reached the output, and deduplicating downstream can't bring them back. Use batch checkpoints where the read and the write happen in the same task, or where re-reading from an earlier point is possible; for end-to-end tracking, use streaming.

Batch checkpoints and streaming

Streaming tracks progress more carefully. Spark keeps the stream's offsets itself, and commits a batch only after the sink has finished it. That tracks the source, though. Whether the output can hold duplicates depends on the sink: a batch replayed after a failure is written again, so a sink that is neither transactional nor idempotent, such as Kafka or a plain foreachBatch, can receive it twice. Deduplicate by batch ID, or write idempotently.

tables {
  events {
    endpoint = "/events"
    checkpoint {
      enabled = true
      path = "/tmp/apilytics/checkpoints"  # a local path, or s3a://, gs://, abfss://, ...
      mode = "cursor"                      # cursor | offset | timestamp
    }
  }
}

`mode`	Resumes from	Saves
`cursor`	The last pagination cursor	The last cursor value
`offset`	The last numeric offset	The next offset
`timestamp`	A record timestamp, sent as a query parameter	The latest record timestamp

Timestamp mode also needs to know which field to track and which parameter to send. checkpoint always belongs to a table:

tables {
  events {
    endpoint = "/events"
    checkpoint {
      enabled = true
      path = "s3a://my-bucket/checkpoints"
      mode = "timestamp"
      timestamp-path = "/updated_at"    # JSON pointer to the timestamp in each record
      timestamp-param = "since"         # query parameter to filter on
    }
  }
}

The checkpoint saves the latest timestamp of the whole read, compared as instants, so the order the API serves pages in doesn't matter.

Every request carries the checkpoint's value in timestamp-param. A filters entry on the same parameter, such as since mapped to updated_at, therefore isn't sent to the API: Spark applies that filter to the rows that come back, and the config logs a warning saying so when it loads. A timestamp-param that pagination also sends, such as its cursor-param, is rejected at load, since only one value could be sent.

A checkpoint can't be combined with partitioning: it's one position for the whole table, so every partition would resume from the same place. A config with both is rejected at load.

Each table's checkpoint is stored as <table-name>.checkpoint.json under path. A plain path, such as /tmp/checkpoints or a Databricks /Volumes/... path, is written directly. A path with a scheme, such as hdfs://, s3a://, gs://, abfss://, wasbs://, dbfs:/ or file://, goes through Hadoop's FileSystem, so the cluster needs the matching connector. A plain local path is on each executor's own disk: on a cluster, use shared storage.

Timestamp mode is also what lets a table stream. A streaming query keeps its offsets in its own checkpointLocation, not in this file.