使用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；跟踪历史在＊除了（城市）