自動cdcフローを作成する

create_auto_cdc_flow()関数は、Lakeflow宣言型パイプラインチェンジデータキャプチャ (CDC) 機能を使用してチェンジデータフィード (CDF) からのソースデータを処理するフローを作成します。

注記

この関数は以前の関数apply_changes()を置き換えます。2 つの関数は同じシグネチャを持ちます。Databricks では、新しい名前を使用するように更新することをお勧めします。

important

変更を適用するターゲットストリーミングテーブルを宣言する必要があります。オプションで、ターゲットテーブルのスキーマを指定できます。create_auto_cdc_flow()ターゲットテーブルのスキーマを指定するときは、 sequence_byフィールドと同じデータ型の__START_AT列と__END_AT列を含める必要があります。

必要なターゲットテーブルを作成するには、Lakeflow 宣言型パイプラインインターフェイスでcreate_streaming_table() Python 関数を使用できます。

構文

Python
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>",
  ignore_null_updates = False,
  apply_as_deletes = None,
  apply_as_truncates = None,
  column_list = None,
  except_column_list = None,
  stored_as_scd_type = <type>,
  track_history_column_list = None,
  track_history_except_column_list = None,
  name = None,
  once = False
)

create_auto_cdc_flow処理の場合、 INSERTおよびUPDATEイベントのデフォルトの動作は、ソースから CDC イベント をアップサートする ことです。つまり、指定されたキーに一致するターゲットテーブルのすべての行を更新するか、一致するレコードがターゲットテーブルに存在しない場合は新しい行を挿入します。DELETEイベントの処理は、 apply_as_deletes問題で指定できます。

変更フィードを使用した CDC 処理の詳細については、「 AUTO CDC APIs: Lakeflow 宣言型パイプラインを使用してチェンジデータキャプチャを簡略化する」を参照してください。 create_auto_cdc_flow() 関数の使用例については、例: CDF ソース・データを使用した SCD タイプ 1 および SCD タイプ 2 の処理を参照してください。

問題

パラメーター	Type	説明
`target`	`str`	必須。更新するテーブルの名前。`create_auto_cdc_flow()`関数を実行する前に、 create_streaming_table()関数を使用してターゲットテーブルを作成できます。
`source`	`str`	必須。CDCレコードを含むデータソース。
`keys`	`list`	必須。ソースデータ内の行を一意に識別する列または列の組み合わせ。これは、どのCDCイベントがターゲットテーブル内の特定のレコードに適用されるかを識別するために使用されます。次のいずれかを指定できます。文字列のリスト： `["userId", "orderId"]` Spark SQL `col()`関数のリスト: `[col("userId"), col("orderId")]` 。`col()`関数の引数には修飾子を含めることはできません。たとえば、 `col(userId)`は使用できますが、 `col(source.userId)`は使用できません。
`sequence_by`	`str`、 `col()`または `struct()`	必須。ソース・データ内の CDC イベントの論理的な順序を指定する列名。Lakeflow 宣言型パイプラインは、このシーケンスを使用して、順不同で到着した変更イベントを処理します。指定する列は、ソート可能なデータ型である必要があります。次のいずれかを指定できます。文字列： `"sequenceNum"` Spark SQL `col()`関数: `col("sequenceNum")` 。`col()`関数の引数には修飾子を含めることはできません。たとえば、 `col(userId)`は使用できますが、 `col(source.userId)`は使用できません。複数の列を結合して同点を解消する`struct()` : `struct("timestamp_col", "id_col")`では、最初に最初の構造体フィールドで順序付けされ、同点の場合は 2 番目のフィールドで順序付けされるなどとなります。
`ignore_null_updates`	`bool`	ターゲットカラムのサブセットを含む更新の取り込みを許可します。 CDC イベントが既存の行と一致し、 `ignore_null_updates`が`True`の場合、 `null`の列はターゲット内の既存の値を保持します。これは、値が`null`のネストされた列にも適用されます。`ignore_null_updates`が`False`の場合、既存の値は`null`値で上書きされます。デフォルトは`False`です。
`apply_as_deletes`	`str` または `expr()`	CDCイベントをupsertではなく`DELETE`として扱う必要がある場合を指定します。次のいずれかを指定できます。文字列： `"Operation = 'DELETE'"` Spark SQL `expr()`関数は次のとおりです： `expr("Operation = 'DELETE'")` 順序が正しくないデータを処理するために、削除された行は基になる Delta テーブルにトゥームストーンとして一時的に保持され、これらのトゥームストーンをフィルター処理するビューがメタストアに作成されます。保持間隔はデフォルトで 2 日に設定されており、 `pipelines.cdc.tombstoneGCThresholdInSeconds`テーブルプロパティを使用して構成できます。
`apply_as_truncates`	`str` または `expr()`	CDCイベントを完全なテーブル`TRUNCATE`として扱う必要がある場合を指定します。次のいずれかを指定できます。文字列： `"Operation = 'TRUNCATE'"` Spark SQL `expr()`関数は次のとおりです： `expr("Operation = 'TRUNCATE'")` この句はターゲットテーブルの完全な切り捨てをトリガーするため、この機能を必要とする特定のユースケースにのみ使用する必要があります。`apply_as_truncates`引数はSCDタイプ 1 でのみサポートされます。SCD タイプ 2 SCD切り捨て操作をサポートしません。
`column_list` または `except_column_list`	`list`	ターゲットテーブルに含める列のサブセット。`column_list`を使用して、含める列の完全なリストを指定します。`except_column_list`を使用して、除外する列を指定します。いずれかの値を文字列のリストとして宣言することも、Spark SQL `col()`関数として宣言することもできます： `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")` `col()`関数の引数には修飾子を含めることはできません。たとえば、 `col(userId)`は使用できますが、 `col(source.userId)`は使用できません。デフォルトでは、関数に`column_list`または`except_column_list`引数が渡されない場合、ターゲットテーブルのすべての列が含まれます。
`stored_as_scd_type`	`str` または `int`	レコードを SCD タイプ 1 として保存するか、SCD タイプ 2 として保存するかを指定します。SCD タイプ 1 の場合は`1` 、SCD タイプ 2 の場合は`2`に設定します。デフォルトは SCD タイプ 1 です。
`track_history_column_list` または `track_history_except_column_list`	`list`	ターゲットテーブル内の履歴を追跡する出力列のサブセット。追跡する列の完全なリストを指定するには、 `track_history_column_list`を使用します。追跡から除外する列を指定するには、 `track_history_except_column_list`を使用します。どちらの値も、文字列のリストまたは Spark SQL `col()`関数として宣言できます。 `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")` `col()`関数の引数には修飾子を含めることはできません。たとえば、 `col(userId)`は使用できますが、 `col(source.userId)`は使用できません。デフォルトでは、関数に`track_history_column_list`または`track_history_except_column_list`引数が渡されない場合、ターゲットテーブルのすべての列が含まれます。
`name`	`str`	フロー名。指定されていない場合は、デフォルトで`target`と同じ値になります。
`once`	`bool`	必要に応じて、フローをバックフィルなどの 1 回限りのフローとして定義します。`once=True`を使用すると、フローは次の 2 つの方法で変化します。戻り値。`streaming-query` 。この場合、ストリーミング DataFrame ではなく、バッチ DataFrame である必要があります。デフォルトでは、フローは 1 回実行されます。パイプラインが完全リフレッシュで更新されると、 `ONCE`フローが再度実行され、データが再作成されます。

構文​

問題​

構文

問題