update_flow

Use the @dp.update_flow decorator to create an update flow. Update flows write to sinks using update output mode, emitting only the rows that changed in each batch. Unlike append flows, they support stateful aggregations without requiring a watermark.

Update flows can only target sinks. Delta tables are not supported.

Update flows can run in real-time mode for sub-second operational workloads. Real-time mode in Lakeflow pipelines supports classic and serverless compute.

Syntax

from pyspark import pipelines as dp

dp.create_sink("<sink-name>", "<format>", {"<key>": "<value>"})

@dp.update_flow(
  target = "<sink-name>",
  name = "<flow-name>", # optional, defaults to function name
  depends_on = "<flow-name>", # optional, Public Preview
  spark_conf = {"<key>" : "<value>", "<key>" : "<value>"}, # optional
  comment = "<comment>") # 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 sink this flow writes to.
name str The flow name. If not provided, defaults to the function name.
depends_on str or list Public Preview. One or more flow names that must complete successfully before this flow starts. Accepts a single flow name or a list of names. In a continuous pipeline, each predecessor must be a one-time (once) flow. This orders flow execution only; it does not change how the flow runs. See Order pipeline flow execution with depends_on.
comment str A description for the flow.
spark_conf dict A dict of Spark configurations for the execution of this query. These configurations override confs set for the destination, pipeline, or cluster.

Examples

Aggregation to a Kafka sink

Write stateful aggregation results to a Kafka sink:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

dp.create_sink("event_counts_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="event_counts_flow",
    target="event_counts_sink",
)
def event_counts():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .selectExpr("CAST(key AS STRING) AS event_type")
            .groupBy(col("event_type"))
            .count()
    )

Real-time mode

Note

Real-time mode requires a pipeline configured for continuous execution. See Use real-time mode in Lakeflow pipelines to check the runtime requirements and enable real-time mode before running this example.

Use spark_conf to configure an update flow for real-time mode:

from pyspark import pipelines as dp

dp.create_sink("my_kafka_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="my_rtm_flow",
    target="my_kafka_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def my_real_time_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
    )

Limitations

  • Delta table sinks are not supported as targets for update flows.
  • Last updated on