DLT表写入Kafka：初始运行正常，后续更新致流任务崩溃

核心原因 #

你的deal_gold1是DLT默认的快照表，支持更新/删除操作。当用readStream读取时，会捕获到这些非追加的变更事件（更新、删除），但Kafka的append输出模式仅能处理新增数据，无法处理变更事件，导致流任务崩溃。

具体解决思路 #

1. 将DLT表改为Append-Only模式（推荐，若业务允许只保留新增数据）

修改DLT表的创建语句，添加delta.appendOnly=true属性，强制表仅接受追加操作，避免产生更新/删除事件：

CREATE OR REFRESH LIVE TABLE deal_gold1
TBLPROPERTIES ("quality" = "gold", "delta.appendOnly" = "true")
COMMENT "Gold Deals (Append-Only)"
AS SELECT
  documentId,
  eventTimestamp,
  substring(fullDocument.owner_id, 11, 24) as owner_id,
  fullDocument.owner_type as owner_type,
  substring(fullDocument.account_id, 11, 24) as account_id,
  substring(fullDocument.manager_account_id, 11, 24) as manager_account_id,
  fullDocument.hubspot_deal_id as hubspot_deal_id,
  fullDocument.stage as stage,
  fullDocument.status as status,
  fullDocument.title as title
FROM LIVE.deal_bronze_cleansed

2. 启用Delta CDC捕获仅新增数据（若需保留DLT表的更新能力）

如果业务需要保留DLT表的更新/删除功能，可通过Delta的变更数据捕获（CDC）读取表的变更日志，仅过滤出新增的记录：

import pyspark.sql.functions as fn
from pyspark.sql.types import StringType

# 读取DLT表的变更日志，仅保留新增记录
df = spark.readStream.format("delta")
    .option("readChangeFeed", "true")
    .table("deal_stream_test.deal_gold1")
    .filter("_change_type = 'insert'")  # 只处理插入事件
    .drop("_change_type", "_commit_version", "_commit_timestamp")  # 移除CDC元数据列

# 后续Kafka写入逻辑
writeStream= (
  df
  .selectExpr("CAST(documentId AS STRING) AS key", "to_json(struct(*)) AS value")
  .writeStream
  .format("kafka")
  .outputMode("append")
  .option("checkpointLocation", "/tmp/benperram21/checkpoint")
  .option("kafka.bootstrap.servers", confluentBootstrapServers)
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.sasl.jaas.config", "kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username='{}' password='{}';".format(confluentApiKey, confluentSecret))
  .option("kafka.ssl.endpoint.identification.algorithm", "https")
  .option("kafka.sasl.mechanism", "PLAIN")
  .option("topic", confluentTopicName)
  .start()
)

3. 清理旧的Checkpoint目录

之前的崩溃可能导致checkpoint目录残留脏数据，建议先删除/tmp/benperram21/checkpoint目录，再重新启动流任务，避免状态不一致问题。

4. 移除无效配置

你的Kafka写入代码中重复设置了ignoreChanges参数，该参数仅适用于Delta表的写入操作，对Kafka写入无意义，应删除该配置。

内容的提问来源于stack exchange，提问作者Ben Perram