ML Engineer MasterClass (October) | 4 seats left

Spark · Checkpoints and recovery
AmazonAmazon Analytics
Amazon · Structured Streaming

Checkpoints and recovery

Resume bounded file streams, verify committed output, and distinguish recovery from replay.

Step 1 of 6 · Learn

Resume from recorded progress

A checkpoint records streaming progress and, for stateful queries, state needed for recovery. This file-to-Parquet example finishes one finite trigger, adds another input file, then restarts the same query with the same checkpoint and destination. The second run should process new files rather than replay completed ones. Trigger.Once is used for these bounded exercises; newer deployments often prefer AvailableNow. Each trigger has a timeout and query cleanup.

Lesson reference: PySpark and Scala

Resume from recorded progress

A checkpoint records streaming progress and, for stateful queries, state needed for recovery. This file-to-Parquet example finishes one finite trigger, adds another input file, then restarts the same query with the same checkpoint and destination. The second run should process new files rather than replay completed ones. Trigger.Once is used for these bounded exercises; newer deployments often prefer AvailableNow. Each trigger has a timeout and query cleanup.

PySpark example

from tempfile import TemporaryDirectory
from pyspark.sql.functions import col

def drain(query):
    try:
        if not query.awaitTermination(45):
            raise RuntimeError("Streaming run timed out; retry the example.")
    finally:
        query.stop()

with TemporaryDirectory(prefix="spark-recovery-") as directory:
    input_path = directory + "/input"
    output_path = directory + "/output"
    checkpoint = directory + "/checkpoint"
    products.filter(col("product_id") <= 203).coalesce(1).write.json(input_path)
    stream = spark.readStream.schema(products.schema).json(input_path)
    first = stream.select("product_id", "product_name")
    drain(first.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(once=True).start())
    products.filter((col("product_id") > 203) & (col("product_id") <= 204)).coalesce(1).write.mode("append").json(input_path)
    # Resume the same query with the same checkpoint and destination.
    resumed = stream.select("product_id", "product_name")
    drain(resumed.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(once=True).start())
    saved = spark.read.parquet(output_path).select("product_id", "product_name")
    # Detach these six-or-fewer records before temporary files are removed.
    result = spark.createDataFrame(saved.collect(), saved.schema).orderBy("product_id")
    result.show(truncate=False)

Scala example

import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import java.nio.file.Files
import org.apache.hadoop.fs.Path

def drain(query: StreamingQuery): Unit = {
    try {
        require(query.awaitTermination(45000L), "Streaming run timed out; retry the example.")
    } finally { query.stop() }
}

val directory = Files.createTempDirectory("spark-recovery-")
val root = new Path(directory.toUri)
val fs = root.getFileSystem(spark.sparkContext.hadoopConfiguration)
val input_path = directory.resolve("input").toString
var output_path = directory.resolve("output").toString
var checkpoint = directory.resolve("checkpoint").toString
val result = try {
    products.filter(col("product_id") <= 203).coalesce(1).write.json(input_path)
    val stream = spark.readStream.schema(products.schema).json(input_path)
    val first = stream.select("product_id", "product_name")
    drain(first.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(Trigger.Once()).start())
    products.filter((col("product_id") > 203) && (col("product_id") <= 204)).coalesce(1).write.mode("append").json(input_path)
    // Resume the same query with the same checkpoint and destination.
    val resumed = stream.select("product_id", "product_name")
    drain(resumed.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(Trigger.Once()).start())
    val saved = spark.read.parquet(output_path).select("product_id", "product_name")
    val output = spark.createDataFrame(spark.sparkContext.parallelize(saved.collect().toSeq), saved.schema).orderBy("product_id")
    output.show(false)
    output
} finally {
    fs.delete(root, true) // Only this example’s unique temporary directory.
}

Restart without new input

A restart is not a command to replay all historical input. With the same compatible query, checkpoint, and file sink, a second finite trigger with no new files should leave committed output unchanged. A checkpoint is not the output dataset or an input backup; recovery depends on the source and sink as well. Do not generalize this example to exactly-once behavior for arbitrary external side effects.

PySpark example

from tempfile import TemporaryDirectory
from pyspark.sql.functions import col

def drain(query):
    try:
        if not query.awaitTermination(45):
            raise RuntimeError("Streaming run timed out; retry the example.")
    finally:
        query.stop()

with TemporaryDirectory(prefix="spark-recovery-") as directory:
    input_path = directory + "/input"
    output_path = directory + "/output"
    checkpoint = directory + "/checkpoint"
    products.filter(col("product_id") <= 203).coalesce(1).write.json(input_path)
    stream = spark.readStream.schema(products.schema).json(input_path)
    first = stream.select("product_id", "product_name")
    drain(first.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(once=True).start())
    # No new files are added before restarting.
    # Resume the same query with the same checkpoint and destination.
    resumed = stream.select("product_id", "product_name")
    drain(resumed.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(once=True).start())
    saved = spark.read.parquet(output_path).select("product_id", "product_name")
    # Detach these six-or-fewer records before temporary files are removed.
    result = spark.createDataFrame(saved.collect(), saved.schema).orderBy("product_id")
    result.show(truncate=False)

Scala example

import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import java.nio.file.Files
import org.apache.hadoop.fs.Path

def drain(query: StreamingQuery): Unit = {
    try {
        require(query.awaitTermination(45000L), "Streaming run timed out; retry the example.")
    } finally { query.stop() }
}

val directory = Files.createTempDirectory("spark-recovery-")
val root = new Path(directory.toUri)
val fs = root.getFileSystem(spark.sparkContext.hadoopConfiguration)
val input_path = directory.resolve("input").toString
var output_path = directory.resolve("output").toString
var checkpoint = directory.resolve("checkpoint").toString
val result = try {
    products.filter(col("product_id") <= 203).coalesce(1).write.json(input_path)
    val stream = spark.readStream.schema(products.schema).json(input_path)
    val first = stream.select("product_id", "product_name")
    drain(first.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(Trigger.Once()).start())
    // No new files are added before restarting.
    // Resume the same query with the same checkpoint and destination.
    val resumed = stream.select("product_id", "product_name")
    drain(resumed.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(Trigger.Once()).start())
    val saved = spark.read.parquet(output_path).select("product_id", "product_name")
    val output = spark.createDataFrame(spark.sparkContext.parallelize(saved.collect().toSeq), saved.schema).orderBy("product_id")
    output.show(false)
    output
} finally {
    fs.delete(root, true) // Only this example’s unique temporary directory.
}

Start fresh for a changed query

Not every query change is compatible with a saved checkpoint. Changes to sources, sink configuration, state schema, or stateful operators may require a new checkpoint; consult the runtime’s recovery rules. Here the second run deliberately uses both a fresh checkpoint and a fresh destination, so it replays all currently available source files without mixing incompatible outputs. The original destination remains untouched until the example’s final cleanup.

PySpark example

from tempfile import TemporaryDirectory
from pyspark.sql.functions import col

def drain(query):
    try:
        if not query.awaitTermination(45):
            raise RuntimeError("Streaming run timed out; retry the example.")
    finally:
        query.stop()

with TemporaryDirectory(prefix="spark-recovery-") as directory:
    input_path = directory + "/input"
    output_path = directory + "/output"
    checkpoint = directory + "/checkpoint"
    products.filter(col("product_id") <= 203).coalesce(1).write.json(input_path)
    stream = spark.readStream.schema(products.schema).json(input_path)
    first = stream.select("product_id", "product_name")
    drain(first.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(once=True).start())
    products.filter((col("product_id") > 203) & (col("product_id") <= 206)).coalesce(1).write.mode("append").json(input_path)
    # A changed output schema uses a fresh checkpoint AND destination.
    output_path = directory + "/fresh_output"
    checkpoint = directory + "/fresh_checkpoint"
    resumed = stream.select("product_id", "product_name")
    drain(resumed.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(once=True).start())
    saved = spark.read.parquet(output_path).select("product_id", "product_name")
    # Detach these six-or-fewer records before temporary files are removed.
    result = spark.createDataFrame(saved.collect(), saved.schema).orderBy("product_id")
    result.show(truncate=False)

Scala example

import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{StreamingQuery, Trigger}
import java.nio.file.Files
import org.apache.hadoop.fs.Path

def drain(query: StreamingQuery): Unit = {
    try {
        require(query.awaitTermination(45000L), "Streaming run timed out; retry the example.")
    } finally { query.stop() }
}

val directory = Files.createTempDirectory("spark-recovery-")
val root = new Path(directory.toUri)
val fs = root.getFileSystem(spark.sparkContext.hadoopConfiguration)
val input_path = directory.resolve("input").toString
var output_path = directory.resolve("output").toString
var checkpoint = directory.resolve("checkpoint").toString
val result = try {
    products.filter(col("product_id") <= 203).coalesce(1).write.json(input_path)
    val stream = spark.readStream.schema(products.schema).json(input_path)
    val first = stream.select("product_id", "product_name")
    drain(first.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(Trigger.Once()).start())
    products.filter((col("product_id") > 203) && (col("product_id") <= 206)).coalesce(1).write.mode("append").json(input_path)
    // A changed output schema uses a fresh checkpoint AND destination.
    output_path = directory.resolve("fresh_output").toString
    checkpoint = directory.resolve("fresh_checkpoint").toString
    val resumed = stream.select("product_id", "product_name")
    drain(resumed.writeStream.format("parquet").outputMode("append")
        .option("path", output_path).option("checkpointLocation", checkpoint)
        .trigger(Trigger.Once()).start())
    val saved = spark.read.parquet(output_path).select("product_id", "product_name")
    val output = spark.createDataFrame(spark.sparkContext.parallelize(saved.collect().toSeq), saved.schema).orderBy("product_id")
    output.show(false)
    output
} finally {
    fs.delete(root, true) // Only this example’s unique temporary directory.
}
example.pyPySpark
1. First finite runRead the initial product files and commit Parquet output with a checkpoint.
2. Resume the existing queryAdd new input files, then reuse both checkpoint and destination.
3. Verify and clean upRead the output records, detach this small result, and remove only the example’s private files.
Source products6 rows
product_idproduct_namecategory
201Wireless keyboardElectronics
202Laptop standOffice
203Desk lampHome
204USB-C hubElectronics
205Travel backpackTravel
206Notebook setOffice
Resume from recorded progress
Result4 rows
product_idproduct_name
201Wireless keyboard
202Laptop stand
203Desk lamp
204USB-C hub
The Learn example produces four products, not seven: the initial three are not appended again during recovery.

The example is loaded in the editor. Run it as written, then try a small change.

Runs on the Spark backend. First startup may take a moment.

Run your code to see the result.