使用Delta Live表更改數據捕獲

應用更改函數

apply_changes（目標=“<目標表>”，源=“<數據源>”，鍵=(“key1”，“key2”，“keyN”]，sequence_by=“< sequence-column >”，ignore_null_updates=假，apply_as_deletes=沒有一個，apply_as_truncates=沒有一個，column_list=沒有一個，except_column_list=沒有一個，stored_as_scd_type=<類型>，track_history_column_list=沒有一個，track_history_except_column_list=沒有一個）

的默認行為插入和更新事件是插入來自源的CDC事件:更新目標表中與指定鍵匹配的任何行，或者在目標表中不存在匹配記錄時插入新行。處理的刪除屬性可以指定事件apply_as_deletes論點。

為輸出記錄創建一個目標表

使用create_streaming_live_table ()方法創建目標表apply_changes ()輸出記錄。

create_streaming_live_table（名字=“<表名稱>”，評論=“< >評論”spark_conf={“< >鍵”：“<價值”，“<鍵”：“< >價值”}，table_properties={“< >鍵”：“< >價值”，“< >鍵”：“< >價值”}，partition_cols=(“<劃分字段>”，“<劃分字段>”]，路徑=“< storage-location-path >”，模式=“模式定義”）

類的模式時apply_changes目標表中，還必須包含__START_AT和__END_AT類具有相同數據類型的列sequence_by字段。例如，如果目標表有列鍵,字符串，值,字符串,測序,長：

create_streaming_live_table（名字=“目標”，評論="目標為疾病控製中心攝入。"，partition_cols=(“價值”]，路徑=“tablePath美元”，模式=StructType（(StructField（“關鍵”，StringType（））,StructField（“價值”，StringType（））,StructField（“排序”，LongType（））,StructField（“__START_AT”，LongType（））,StructField（“__END_AT”，LongType（))］））

生成測試數據

為這個例子創建測試記錄:

1. 轉到Databricks登錄頁並選擇創建一個筆記本，或按新在側欄中選擇筆記本。的創建筆記本對話框出現了。

2. 在創建筆記本對話，給你的筆記本起個名字;例如,生成測試CDC記錄。選擇SQL從默認的語言下拉菜單。

3. 如果有正在運行的集群，則集群下拉顯示。選擇要將筆記本附加到的集群。你也可以創建創建筆記本後要附加到的新集群。

4. 點擊創建。

5. 複製以下查詢並將其粘貼到新筆記本的第一個單元格中:

   創建模式如果不存在cdc_data；創建表格cdc_data。用戶作為選擇col1作為用戶標識，col2作為名字，col3作為城市，col4作為操作，col5作為sequenceNum從（值——初始負載。（124，“勞爾”，“瓦哈卡”，“插入”，1)，（123，“伊莎貝爾”，“蒙特雷”，“插入”，1)，——新用戶。（125，“梅賽德斯”，“提華納”，“插入”，2)，（126，“莉莉”，“坎昆”，“插入”，2)，伊莎貝爾被從係統中移除，梅賽德斯被轉移到瓜達拉哈拉。（123，零，零，“刪除”，6)，（125，“梅賽德斯”，“瓜達拉哈拉”，“更新”，6)，—這批更新是不按順序來的。上述位於sequenceNum 5的批處理將是最終狀態。（125，“梅賽德斯”，“墨西卡利”，“更新”，5)，（123，“伊莎貝爾”，“吉娃娃”，“更新”，5）—取消注釋以測試TRUNCATE。——，(null, null, null， "TRUNCATE"， 3)）;

6. 要運行筆記本並填充測試記錄，請在單元格操作菜單中在最右側，單擊並選擇運行單元，或按shift + enter。

創建並運行SCD類型1示例管道

1. 轉到Databricks登錄頁並選擇創建一個筆記本，或按新在側欄中選擇筆記本。的創建筆記本對話框出現了。

2. 在創建筆記本對話，給你的筆記本起個名字;例如,DLT CDC示例。選擇Python或SQL從默認的語言下拉菜單，根據您的首選語言。你可以離開了集群設置為默認值。Delta Live Tables運行時在運行管道之前創建一個集群。

3. 點擊創建。

4. 複製Python或SQL查詢然後粘貼到筆記本的第一個單元格中。

5. 創建一個新的管道並將筆記本添加到筆記本庫字段。要發布管道處理的輸出，可以在目標字段。

6. 啟動管道。如果您配置了目標價值，你可以查看並驗證結果查詢的。

示例查詢

進口dlt從pyspark.sql.functions進口上校，expr@dlt。視圖def用戶（）:返回火花。readStream。格式（“δ”）。表格（“cdc_data.users”）dlt。create_streaming_live_table（“目標”）dlt。apply_changes（目標=“目標”，源=“用戶”，鍵=(“標識”]，sequence_by=上校（“sequenceNum”)，apply_as_deletes=expr（"operation = 'DELETE'")，apply_as_truncates=expr（"操作= 'TRUNCATE'")，except_column_list=(“操作”，“sequenceNum”]，stored_as_scd_type=1）

——創建並填充目標表。創建或刷新流媒體生活表格目標；應用變化成生活。目標從流（cdc_data。用戶）鍵（用戶標識）應用作為刪除當操作=“刪除”應用作為截斷當操作=“截斷”序列通過sequenceNum列*除了（操作，sequenceNum）存儲作為鏡頭分割類型1；

創建並運行SCD類型2示例管道

1. 轉到Databricks登錄頁並選擇創建一個筆記本，或按新在側欄中選擇筆記本。的創建筆記本對話框出現了。

2. 在創建筆記本對話，給你的筆記本起個名字;例如,DLT CDC示例。選擇Python或SQL從默認的語言下拉菜單，根據您的首選語言。你可以離開了集群設置為默認值。Delta Live Tables運行時在運行管道之前創建一個集群。

3. 點擊創建。

4. 複製Python或SQL查詢然後粘貼到筆記本的第一個單元格中。

5. 創建一個新的管道並將筆記本添加到筆記本庫字段。要發布管道處理的輸出，可以在目標字段。

6. 啟動管道。如果您配置了目標價值，你可以查看並驗證結果查詢的。

示例查詢

進口dlt從pyspark.sql.functions進口上校，expr@dlt。視圖def用戶（）:返回火花。readStream。格式（“δ”）。表格（“cdc_data.users”）dlt。create_streaming_live_table（“目標”）dlt。apply_changes（目標=“目標”，源=“用戶”，鍵=(“標識”]，sequence_by=上校（“sequenceNum”)，apply_as_deletes=expr（"operation = 'DELETE'")，except_column_list=(“操作”，“sequenceNum”]，stored_as_scd_type=“2”）

——創建並填充目標表。創建或刷新流媒體生活表格目標；應用變化成生活。目標從流（cdc_data。用戶）鍵（用戶標識）應用作為刪除當操作=“刪除”序列通過sequenceNum列*除了（操作，sequenceNum）存儲作為鏡頭分割類型2；

帶有跟蹤曆史的示例查詢

進口dlt從pyspark.sql.functions進口上校，expr@dlt。視圖def用戶（）:返回火花。readStream。格式（“δ”）。表格（“cdc_data.users”）dlt。create_streaming_live_table（“目標”）dlt。apply_changes（目標=“目標”，源=“用戶”，鍵=(“標識”]，sequence_by=上校（“sequenceNum”)，apply_as_deletes=expr（"operation = 'DELETE'")，except_column_list=(“操作”，“sequenceNum”]，stored_as_scd_type=“2”，track_history_except_column_list=(“城市”］）

——創建並填充目標表。創建或刷新流媒體生活表格目標；應用變化成生活。目標從流（cdc_data。用戶）鍵（用戶標識）應用作為刪除當操作=“刪除”序列通過sequenceNum列*除了（操作，sequenceNum）存儲作為鏡頭分割類型2；跟蹤曆史在*除了（城市）