create_auto_cdc_flow

The create_auto_cdc_flow() function creates a flow that uses Lakeflow pipelines change data capture (CDC) functionality to process source data from a change data feed (CDF).

Note

This function replaces the previous function apply_changes(). The two functions have the same signature. Databricks recommends updating to use the new name.

Important

You must declare a target streaming table to apply changes into. You can optionally specify the schema for your target table. When specifying the schema of the create_auto_cdc_flow() target table, you must include the __START_AT and __END_AT columns with the same data type as the sequence_by fields.

To create the required target table, you can use the create_streaming_table() function in the pipeline Python interface.

Syntax

from pyspark import pipelines as dp

dp.create_auto_cdc_flow(
  target = "<target-table>",
  source = "<data-source>",
  keys = ["key1", "key2", "keyN"],
  sequence_by = "<sequence-column>",
  system_sequence_by = None, # optional
  ignore_null_updates = False, # optional
  ignore_null_updates_column_list = None, # optional
  ignore_null_updates_except_column_list = None, # optional
  columns_to_update = None, # optional
  apply_as_deletes = None, # optional
  apply_as_truncates = None, # optional
  column_list = None, # optional
  except_column_list = None, # optional
  stored_as_scd_type = "1", # optional
  track_history_column_list = None, # optional
  track_history_except_column_list = None, # optional
  name = None, # optional
  once = False # optional
)

For create_auto_cdc_flow processing, the default behavior for INSERT and UPDATE events is to upsert CDC events from the source: update any rows in the target table that match the specified key(s) or insert a new row when a matching record does not exist in the target table. Handling for DELETE events can be specified with the apply_as_deletes parameter.

To learn more about CDC processing with a change feed, see The AUTO CDC APIs: Simplify change data capture with pipelines. For an example of using the create_auto_cdc_flow() function, see AUTO CDC examples.

Parameters

Parameter	Type	Description
target	str	Required. The name of the table to be updated. You can use the create_streaming_table() function to create the target table before executing the create_auto_cdc_flow() function.
source	str	Required. The data source containing CDC records.
keys	list	Required. The column or combination of columns that uniquely identify a row in the source data. This is used to identify which CDC events apply to specific records in the target table. You can specify either: A list of strings: ["userId", "orderId"] A list of Spark SQL col() functions: [col("userId"), col("orderId")]. Arguments to col() functions cannot include qualifiers. For example, you can use col(userId), but you cannot use col(source.userId).
sequence_by	str, col() or struct()	Required. The column names specifying the logical order of CDC events in the source data. Lakeflow pipelines use this sequencing to handle change events that arrive out of order. The specified column must be a sortable data type. You can specify either: A string: "sequenceNum" A Spark SQL col() function: col("sequenceNum"). Arguments to col() functions cannot include qualifiers. For example, you can use col(userId), but you cannot use col(source.userId). A struct() combining multiple columns to break ties: struct("timestamp_col", "id_col"), it orders by the first struct field first, then by the second field if there's a tie, and so on.
system_sequence_by	str or col()	The column specifying the system time at which each CDC event is known to the system. Used with stored_as_scd_type="bitemporal" to track changes across both business time (sequence_by) and system time. The specified column must be a sortable data type. Bitemporal AUTO CDC is in Beta. See Bitemporal AUTO CDC. This parameter is optional and applies only to bitemporal tables.
ignore_null_updates	bool	Controls how null values in incoming CDC updates are handled. When ignore_null_updates is True, null columns in an incoming update are ignored; the existing value in the target row is preserved. This also applies to nested columns with null values. When ignore_null_updates is False, null columns in an incoming update overwrite existing values in the target. Defaults to False. Set to True when source events only include the columns that changed, so unchanged columns are not overwritten with null. The default is False. For finer control over which columns ignore null values, use ignore_null_updates_column_list or ignore_null_updates_except_column_list.
ignore_null_updates_column_list	list	A subset of columns for which null values in an incoming change record are ignored, so each of those columns keeps its existing value in the target. Columns outside the list apply explicit null values. Use this parameter to apply partial updates when your source sends only the columns that changed. Equivalent to the SQL IGNORE NULL UPDATES ON columnList clause. Use either ignore_null_updates_column_list or ignore_null_updates_except_column_list, not both.
ignore_null_updates_except_column_list	list	A subset of columns that apply explicit null values. All other columns ignore null values in an incoming change record and keep their existing value in the target. Equivalent to the SQL IGNORE NULL UPDATES ON * EXCEPT (...) clause. Use either ignore_null_updates_column_list or ignore_null_updates_except_column_list, not both.
columns_to_update	str or col()	The name of a source column that holds, for each change record, the set of columns to update as an array of column-name strings (array<string>). Columns not in the array keep their existing target values, while listed columns are written from the source, including explicit null values. Use this parameter when each change record updates a different set of columns and you need to apply explicit null values. Equivalent to the SQL COLUMNS TO UPDATE clause. You cannot set columns_to_update together with ignore_null_updates, ignore_null_updates_column_list, or ignore_null_updates_except_column_list. columns_to_update is not supported for bitemporal tables.
apply_as_deletes	str or expr()	Specifies when a CDC event should be treated as a DELETE rather than an upsert. You can specify either: A string: "Operation = 'DELETE'" A Spark SQL expr() function: expr("Operation = 'DELETE'") To handle out-of-order data, the deleted row is temporarily retained as a tombstone in the underlying Delta table, and a view is created in the metastore that filters out these tombstones. The retention interval defaults to two days, and can be configured with the pipelines.cdc.tombstoneGCThresholdInSeconds table property. If you use Auto Loader as the source for your CDC pipeline, Auto Loader does not guarantee file processing order. For details, see Handle out-of-order data. Set pipelines.cdc.tombstoneGCThresholdInSeconds to a value that exceeds the maximum expected delay between event arrival and pipeline execution. This ensures that delete tombstones are retained long enough to correctly handle late-arriving or out-of-order deletion events.
apply_as_truncates	str or expr()	Specifies when a CDC event should be treated as a full table TRUNCATE. You can specify either: A string: "Operation = 'TRUNCATE'" A Spark SQL expr() function: expr("Operation = 'TRUNCATE'") Because this clause triggers a full truncate of the target table, it should be used only for specific use cases requiring this functionality. The apply_as_truncates parameter is supported only for SCD type 1. SCD type 2 does not support truncate operations.
column_list or except_column_list	list	A subset of columns to include in the target table. Use column_list to specify the complete list of columns to include. Use except_column_list to specify the columns to exclude. You can declare either value as a list of strings or as Spark SQL col() functions: column_list = ["userId", "name", "city"] column_list = [col("userId"), col("name"), col("city")] except_column_list = ["operation", "sequenceNum"] except_column_list = [col("operation"), col("sequenceNum") Arguments to col() functions cannot include qualifiers. For example, you can use col(userId), but you cannot use col(source.userId). The default is to include all columns in the target table when no column_list or except_column_list argument is passed to the function.
stored_as_scd_type	str or int	Whether to store records as SCD type 1, SCD type 2, or bitemporal. Set to 1 for SCD type 1, 2 for SCD type 2, or "bitemporal" to track changes across both business time and system time. Bitemporal requires system_sequence_by and is in Beta. See Bitemporal AUTO CDC. The default is SCD type 1.
track_history_column_list or track_history_except_column_list	list	A subset of output columns to be tracked for history in the target table. Use track_history_column_list to specify the complete list of columns to be tracked. Use track_history_except_column_list to specify the columns to be excluded from tracking. You can declare either value as a list of strings or as Spark SQL col() functions: track_history_column_list = ["userId", "name", "city"] track_history_column_list = [col("userId"), col("name"), col("city")] track_history_except_column_list = ["operation", "sequenceNum"] track_history_except_column_list = [col("operation"), col("sequenceNum") Arguments to col() functions cannot include qualifiers. For example, you can use col(userId), but you cannot use col(source.userId). The default is to include all columns in the target table when no track_history_column_list or track_history_except_column_list argument is passed to the function.
name	str	The flow name. If not provided, defaults to the same value as target.
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.