append_flow

The @dp.append_flow decorator creates append flows or backfills for your pipeline tables. The function must return an Apache Spark streaming DataFrame. See Load and process data incrementally with Lakeflow pipeline flows.

Append flows can target streaming tables or sinks.

Syntax

from pyspark import pipelines as dp

dp.create_streaming_table("<target-table-name>") # Required only if the target table doesn't exist.

@dp.append_flow(
  target = "<target-table-name>",
  name = "<flow-name>", # optional, defaults to function name
  once = False, # optional
  spark_conf = {"<key>" : "<value", "<key" : "<value>"}, # optional
  comment = "<comment>", # optional
  import_checkpoint = "<checkpoint-path>") # optional
def <function-name>():
  return (<streaming-query>) #

Parameters

Parameter Type Description
function function Required. A function that returns an Apache Spark streaming DataFrame from a user-defined query.
target str Required. The name of the table or sink that is the target of the append flow.
name str The flow name. If not provided, defaults to the function name.
once bool Optionally, define the flow as a one-time flow, such as a backfill. Using once=True changes the flow in two ways:
  • The return value. streaming-query. must be a batch DataFrame in this case, not a streaming DataFrame.
  • The flow is run one time by default. If the pipeline is updated with a complete refresh, then the ONCE flow runs again to recreate the data.
comment str A description for the flow.
spark_conf dict A list of Spark configurations for the execution of this query
import_checkpoint str The path to an existing Structured Streaming checkpoint to import into the flow, so a migrated stream resumes from its last committed offset instead of reprocessing the source. Importing a checkpoint is in Beta. See Migrate a Structured Streaming checkpoint.

Examples

from pyspark import pipelines as dp

# Create a sink for an external Delta table
dp.create_sink("my_sink", "delta", {"path": "/tmp/delta_sink"})

# Add an append flow to an external Delta table
@dp.append_flow(name = "flow", target = "my_sink")
def flowFunc():
  return <streaming-query>

# Add a backfill
@dp.append_flow(name = "backfill", target = "my_sink", once = True)
def backfillFlowFunc():
    return (
      spark.read
      .format("json")
      .load("/path/to/backfill/")
    )

# Create a Kafka sink
dp.create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

# Add an append flow to a Kafka sink
@dp.append_flow(name = "flow", target = "my_kafka_sink")
def myFlow():
  return read_stream("xxx").select(F.to_json(F.struct("*")).alias("value"))

Migrate a Structured Streaming checkpoint

Important

Importing a checkpoint is in Beta.

Use import_checkpoint to migrate an existing Structured Streaming workload to a pipeline without reprocessing the source. Set it to the checkpointLocation your Structured Streaming query used, which can be a cloud storage, Unity Catalog volume, or DBFS path. On the first pipeline update, the flow clones that checkpoint into the pipeline's managed storage. The flow then resumes from the last committed offset with its state (such as aggregations, deduplication keys, and watermarks) intact. Subsequent pipeline updates use the flow's cloned checkpoint; the original checkpoint is not modified.

The flow must target a managed table created with create_table or a sink.

Stop the original Structured Streaming query before you run the pipeline. The original Structured Streaming query can be reused after the import, but you need to manage its checkpoint state and make sure the pipeline and the query do not write to the same table at the same time, which can produce duplicate data.

Recreate the Structured Streaming query as a pipeline flow that writes to a new table and imports its checkpoint:

from pyspark import pipelines as dp

# Create a new managed table for the pipeline
dp.create_table("target_table")

# Continue from the imported checkpoint instead of reprocessing the source.
@dp.append_flow(
  target = "target_table",
  import_checkpoint = "/Volumes/my_catalog/my_schema/checkpoints/my_stream",
)
def migrate():
  # The same source your original query read from.
  return spark.readStream.table("source_table")

The checkpoint is imported only once, on the first pipeline update; later updates ignore import_checkpoint. A full refresh does not re-import the checkpoint; it starts from a new, empty checkpoint and reprocesses the source. To import a different checkpoint, use a flow name that has not been used for the target table before; reusing an existing flow name skips the import.

Limitations

  • Importing a checkpoint into a table that already exists (for example, the original Structured Streaming query target) is not supported. Target a new table that the pipeline creates, or a sink.
  • import_checkpoint is supported only on append_flow.
  • Last updated on