create_auto_cdc_from_snapshot_flow

該 create_auto_cdc_from_snapshot_flow 函式建立一個流程，利用 Lakeflow pipelines 變更資料擷取（CDC）功能來處理資料庫快照中的來源資料。 請參考 SNAPSHOT 中的自動 CDC 運作方式。

備註

此函數取代了先前的函數 apply_changes_from_snapshot()。 這兩個函數具有相同的簽章。 Databricks 建議更新以使用新名稱。

這很重要

您必須具備一個目標串流資料表才能進行此操作。 若要建立所需的目標表格，您可以使用 create_streaming_table（） 函數。

語法

from pyspark import pipelines as dp

dp.create_auto_cdc_from_snapshot_flow(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = "1",
  track_history_column_list = None,
  track_history_except_column_list = None,
  once = False
)

備註

對於 AUTO CDC FROM SNAPSHOT 處理，預設行為是在目標中不存在具有相同索引鍵的相符記錄時插入新列。 如果相符記錄確實存在，則只有在列中的任何值已變更時才會更新該記錄。 目標中存在索引鍵但來源中不再存在的資料列會被刪除。

若要深入瞭解使用快照集進行 CDC 處理，請參閱 自動 CDC API：使用管線簡化變更資料擷取。 關於使用此 create_auto_cdc_from_snapshot_flow() 函數的範例，請參閱 週期快照擷取、 歷史快照擷取及 SCD Type 1 回填 範例。

參數

參數 類型 Description
target str 必須的。 要更新的資料表名稱。 您可以在執行函數之前使用create_auto_cdc_from_snapshot_flow()函數來建立目標表。
source str 或 lambda function 必須的。 要定期快照的資料表或檢視的名稱，或是一個 Python lambda 函數，該函數會傳回要處理的快照資料框和快照版本。 請參閱 實作 source 引數。
keys list 必須的。 唯一定義來源資料中資料列的欄或欄組合。 這可用來識別哪些 CDC 事件套用至目標資料表中的特定記錄。 您可以指定下列其中一項：
  • 字串清單： ["userId", "orderId"]
  • Spark SQL col() 函數清單： [col("userId"), col("orderId"]。 函數的 col() 引數不能包含限定詞。 例如，您可以使用 col(userId)，但不能使用 col(source.userId)。
stored_as_scd_type str 或 int 是否要將記錄儲存為 SCD 類型 1 或 SCD 類型 2。 設為1 以用於 SCD 類型 1，或設為2 以用於 SCD 類型 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 參數給函數時，預設為包含目標表中的所有資料行。
once bool 是否只執行快照流程一次。 當 True時，流程在成功提交後的增量更新中會被跳過。 目標的完整刷新會重新執行流程。 使用此選項進行一次性快照載入或回填。 預設值為 False。

Notes

對於觸發管線中的 SCD Type 1 目標，一個create_auto_cdc_from_snapshot_flow()流程可以與一個或多個AUTO CDC以 Python 或 SQL 定義的流程共用一個目標。 下列為適用需求：

  • 給每個 AUTO CDC 流程一個獨特的名稱。
  • 所有流程都用相同數量、順序相同的鍵。 快照流程的鍵名會與 AUTO CDC 鍵名進行不區分大小寫的比較。 多個 AUTO CDC 流程必須使用相同的鍵名和套管。
  • 快照版本和每個 AUTO CDC 流程的序列欄位，使用完全相同的資料型態。
  • 不要對流程設定期望 AUTO CDC FROM SNAPSHOT 。
  • 請勿使用 IGNORE NULL UPDATES。 在 Python 中，不要在流程中設定 ignore_null_updates、 ignore_null_updates_column_list，或 create_auto_cdc_flow()ignore_null_updates_except_column_list 。
  • 如果你在現有AUTO CDC FROM SNAPSHOT目標中新增AUTO CDC流程，該目標不得包含與保留AUTO CDC系統欄位名稱衝突的使用者欄位。
  • 此圖樣不適用於SCD第二型或雙側顳葉靶標。

舉例請參見 AUTO CDC SCD TYPE 1 表格新增回填。

實作 source 引數

函數create_auto_cdc_from_snapshot_flow()包含引數source。 為了處理歷史快照， source 引數應該是一個 Python lambda 函數，該函數會向函數 create_auto_cdc_from_snapshot_flow() 傳回兩個值：包含要處理的快照資料的 Python DataFrame 和快照版本。

函式的第一次呼叫必須回傳快照資料框及其版本。 如果第一次調用回傳 None，管線更新將失敗。 處理至少一個快照後，回傳 None 表示沒有額外的快照可用。

以下是 lambda 函數的簽名：

lambda Any => Optional[(DataFrame, Any)]
  • lambda 函數的引數是最新處理的快照版本。
  • lambda 函數的傳回值是 None 或兩個值的元組：元組的第一個值是包含要處理的快照的 DataFrame。 元組的第二個值是快照版本，代表快照的邏輯順序。

實作和呼叫 lambda 函數的範例：

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

create_auto_cdc_from_snapshot_flow(
  # ...
  source = next_snapshot_and_version,
  # ...
)

Lakeflow 管線執行時，每次觸發包含該 create_auto_cdc_from_snapshot_flow() 函式的管線時，會執行以下步驟：

  1. 執行 next_snapshot_and_version 函式以載入下一個快照集 DataFrame 和對應的快照版本。
  2. 若第一次呼叫未回傳 DataFrame，則更新失敗。 若後續調用未回傳 DataFrame，執行結束，管線更新標記為完成。
  3. 偵測新快照中的變更，並以累加方式將它們套用至目標資料表。
  4. 返回步驟 #1 以載入下一個快照及其版本。
  • 上次更新時間：