Checkpoints and Dedupe

Checkpoints and Dedupe

Checkpoints

out = job.run(
    df,
    backend="spark",                      # or "local" / "multiprocessing"
    checkpoint="/Volumes/main/ml/checkpoints/ticket_routing",
    id_col="ticket_id",
)

Results are persisted while the run progresses. When a run is interrupted (cluster restart, spot loss, Ctrl-C, quota exhausted), the same call continues where it stopped. Rows already stored are not sent to the model again.

Rules

• id_col must hold unique strings or integers. This is checked first: missing ids, other types and duplicates are a ConfigError.
• A checkpoint directory belongs to one job. Its manifest stores the job's fingerprint(), which covers everything that changes results: model identity, task, prompt/system/text/audio settings, output schema, sampling, engine. A different job on the same directory is a ConfigError. Use a new directory when you change the prompt. Options that don't change results (engine_options, copy_to_local, stage_dir, api_key_env) can change between runs.
• Rows stored with an error are processed again on the next run (retry_failed=True, default). With retry_failed=False they are returned as stored.
• Per id, a stored success wins over a stored failure, then the newest result.
• The input may grow between runs: only new and failed ids are processed.

Local and multiprocessing

Directory layout: manifest.json + results/part-*.jsonl (one JSON object per row: id, value, error, meta). Finished batches are buffered and flushed every checkpoint_every rows (default 1 000), at the end, and on interrupts. A hard kill loses at most one buffer.

Spark

With a checkpoint, the Spark run is eager:

1. Check id_col uniqueness (one aggregation).
2. Skip ids already stored (anti-join).
3. Process the rest in waves of checkpoint_every rows (default 10 000). Each wave is one Spark job whose results are appended to <checkpoint>/results (parquet, typed output columns + _written_at).
4. Return the input left-joined with the stored results.

The returned DataFrame only reads parquet: further actions don't run inference again. A failure loses at most the wave in progress. Smaller waves mean finer-grained progress, at a few seconds of Spark overhead per wave.

The checkpoint path is any path Spark can read and write: /Volumes/..., dbfs:/..., s3://..., abfss://..., or a local path on single-node clusters.

Inspecting a checkpoint

# Spark
spark.read.parquet("/Volumes/main/ml/checkpoints/ticket_routing/results").groupBy(
    F.col("output_error").isNull()
).count().show()

# local
import pandas as pd, glob
pd.concat(pd.read_json(f, lines=True) for f in glob.glob("ckpt/results/*.jsonl"))

Dedupe

job.run(df, dedupe=True)

Rows with identical values in the job's input columns (the template's columns, or the audio column) are sent to the model once. The other rows get the same value or error. Their meta has attempts=0 and no tokens or latency, so statistics and costs count the call once and RunStats.reused reports how many rows were served this way.

• local / multiprocessing: across the whole input (of the rows still to process, with a checkpoint).
• spark: within each Arrow batch (no extra shuffle). For global dedupe on Spark, run the job on df.select(cols).distinct() and join the results back.

Dedupe is only safe when identical inputs should produce identical outputs, which is true for temperature=0, embeddings and transcription. With sampling you trade diversity for cost.