A Retried File Should Not Become New Training Data

Botie’s loader finishes a sensor file and the connection fails before the interface sees its success. The scheduler retries the job. Appending the same file again makes the training dataset larger, even though no new observations arrived.

We need a durable answer to which files have already contributed. We also need that answer to agree with the output: recording progress before the rows are safely written can lose data instead of duplicating it.

A scheduler retries after losing the acknowledgement. Botie must not learn from a second copy of the same file.

{
  "rows": [
    {
      "id": "r0"
    },
    {
      "id": "r1"
    }
  ],
  "checkpoint": [
    "f0",
    "f1"
  ]
}

Imagine a short list of files waiting to be loaded. Each file has an identity and a list of sensor records. The checkpoint keeps the identities whose rows have already reached the committed output. That is the information a restarted job needs before deciding what remains to do.

Our Python representation mirrors those pieces: a list of file dictionaries, a rows list inside each file, and a checkpoint list. A set is useful for asking whether an identity is already present, while the returned list preserves a deliberate order.

In this example, one identity always refers to the same contents. If a producer overwrites a path with new data, we need a version or a different replacement policy before the same rule is safe.

Two different files can still contain the same business event. Detecting that requires an event identity as well as file progress.

completed = {"f0", "f2"}
print("f0" not in completed)
print("f1" not in completed)

False
True

A notification tells us that a file may be available. It does not prove the file has contributed to the output. Select discoveries absent from committed progress, and suppress identical repeated discoveries in the current batch.

If one batch presents different contents under the same immutable file identity, stop rather than arbitrarily taking the first. The conflict is evidence that the identity contract is broken. The checkpoint alone cannot compare a new file with historical contents that were never retained in this model.

The accepted-file preview lets us review what the next commit will include.

There are two ways to leave a misleading recovery record. We could write the rows and fail before recording that their file completed. A retry would append them again. Or we could record completion before the rows are safely stored. A retry would skip a file whose output is missing.

Our commit component returns both changes as one new state, leaving the earlier state untouched. That is the relationship we want from an atomic update: the rows and their progress agree about what completed.

A real destination, often called the sink, needs a durable transaction or an operation identity that makes the same write have one effect when repeated. That repeat-safe property is called idempotency. Returning two Python fields together helps us inspect the rule; real storage must enforce it through failures.

earlier checkpoint = [f0]
new accepted file identities = [f1]
next checkpoint = [f0, f1]
retry with this checkpoint: accepted file identities = []

On a successful replay, the committed dataset should be the same even when the scheduler has attempted the job more than once.

The reference uses a fixed schema and a persistent checkpoint path, then writes to a Delta table. availableNow asks the query to process the available input and finish; awaitTermination waits for that query. Keep the checkpoint associated with this logical query.

Choose source, checkpoint and output locations deliberately. Deleting a checkpoint is a state change with replay consequences. It is not a harmless way to clear an error. Changing the target or reusing a checkpoint for another query requires a reviewed recovery plan.

Lakeflow can manage checkpointing for its own pipelines. This standalone reference makes the path visible because checkpoint identity is what we are learning.

source_path = "/Volumes/lab/training/landing/sensors"
checkpoint_path = "/Volumes/lab/training/checkpoints/sensor_ingest"
query = (spark.readStream.format("cloudFiles")
         .option("cloudFiles.format", "json")
         .schema("pump STRING, temperature DOUBLE")
         .load(source_path)
         .writeStream
         .option("checkpointLocation", checkpoint_path)
         .trigger(availableNow=True)
         .toTable("lab.training.raw_readings"))
query.awaitTermination()

Sending an email, charging a customer or calling a tool during a streaming batch is another effect. The table transaction cannot automatically make an arbitrary remote API call exactly once. Give such effects their own durable operation identities and recovery procedure.

Even a valid file checkpoint does not decide how to handle late events, corrected records or schema drift. Those require their own declared contracts. We are adding one reliable boundary to the pipeline, not declaring every later operation safe.

The next lesson gives corrected entity records an explicit source sequence so arrival order cannot choose the current truth.

A useful test runs the accepted result back through the same discoveries. Nothing should be appended. Another test supplies a new file identity and checks that its rows do arrive. Include an empty file: it can still be a completed input even though it contributes no rows.

Keep old input structures unchanged in these tests. Otherwise a successful assertion can depend on the function having quietly altered the checkpoint or removed rows from the discovery fixture.

Try an empty file as well as a replay, and check the checkpoint alongside the output rows.