replace_flow

Important

This feature is in Beta.

The @dp.replace_flow decorator creates a REPLACE USING flow for a streaming table in your pipeline. On each update, the flow replaces all rows in the target table that match the replace_using key columns and leaves all other rows untouched. The function must return an Apache Spark streaming DataFrame. See Partial snapshot replacement with REPLACE USING flows.

Use @dp.replace_flow when your source is a series of partial snapshots keyed by column. To define the target table and the flow in a single statement instead, pass replace_using and sequence_by to @dp.table.

Syntax

from pyspark import pipelines as dp

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

@dp.replace_flow(
  target = "<target-table-name>",
  replace_using = ["<key-column>", "<key-column>"],
  sequence_by = "<sequence-column>",
  name = "<flow-name>", # optional, defaults to function name
  comment = "<comment>", # optional
  spark_conf = {"<key>" : "<value>", "<key>" : "<value>"}) # 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 streaming table that is the target of the flow.
replace_using	list	Required. The key columns that identify which target rows to replace. Specify at least one column. Key columns can't be repeated, and each key column's type must be sortable.
sequence_by	str or Column	Required. The column that orders the updates. For each key, the highest sequence wins, and a lower-sequence row never overwrites a higher one already in the target.
name	str	The flow name. If not provided, defaults to the function name.
comment	str	A description for the flow.
spark_conf	dict	A list of Spark configurations for the execution of this query.

Examples

from pyspark import pipelines as dp

# Keep the latest row for each order from a stream of partial snapshots
dp.create_streaming_table("orders_current")

@dp.replace_flow(
  target = "orders_current",
  replace_using = ["order_id"],
  sequence_by = "updated_at"
)
def orders_flow():
  return spark.readStream.table("order_updates")

Use more than one key column when a record is identified by a combination of columns:

from pyspark import pipelines as dp

dp.create_streaming_table("accounts_current")

@dp.replace_flow(
  target = "accounts_current",
  replace_using = ["region", "account_id"],
  sequence_by = "updated_at"
)
def accounts_flow():
  return spark.readStream.table("account_updates")

Limitations

• A streaming table supports a single REPLACE USING flow, and can't combine REPLACE USING with another flow type such as an append flow, an auto CDC flow, or a REPLACE WHERE flow.
• The query must be a streaming query. @dp.replace_flow rejects a non-streaming source.
• REPLACE USING flows require Databricks Runtime 18.2 and above.