无法将Kafka数据写入Databricks的Delta Live Table求助

针对你遇到的DLT Pipeline无法加载Kafka数据（但Notebook可正常读取）的问题，结合Unity Catalog环境，从以下几个方向排查：

1. 权限与身份差异 #

• DLT Pipeline默认使用服务主体运行，而Notebook用的是你的个人用户身份，二者权限可能不一致：
  • 验证服务主体是否有Kafka集群的访问权限：如果Kafka启用了SASL/SSL认证，确保服务主体拥有对应的凭证（如密钥、证书），且有权限订阅topic1。
  • 检查服务主体对Unity Catalog的权限：确认其对目标数据库、目录有CREATE TABLE和INSERT权限，若使用UC Secrets存储Kafka配置，需确保服务主体能读取对应Secret。

2. Kafka配置完整性 #

• 代码中{Masked}的kafka.bootstrap.servers配置，在Pipeline中可能未正确传递：
  • 若使用UC Secrets存储地址，需显式通过dbutils.secrets.get读取，而非硬编码或变量引用（Notebook的变量可能未同步到Pipeline）：

    .option("kafka.bootstrap.servers", dbutils.secrets.get("your-secret-scope", "kafka-bootstrap-servers"))
    

  • 补充Kafka安全配置：如果Notebook集群默认继承了SSL/SASL配置，Pipeline集群可能没有，需在代码中显式添加，例如：

    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("kafka.sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username='xxx' password='xxx';")
    

3. 集群与网络配置 #

• 确认DLT Pipeline使用的集群版本：确保集群的Kafka连接器版本与目标Kafka集群版本兼容（如Kafka 2.8+对应连接器版本需匹配）。
• 检查网络连通性：Pipeline集群需能访问Kafka的bootstrap服务器端口（默认9092/9093），确认VPC peering、安全组规则是否开放对应端口。

4. 数据解析与错误处理 #

• 即使Notebook能解析数据，Pipeline运行时可能因Schema不匹配导致数据被丢弃，添加容错解析逻辑：

  from pyspark.sql.functions import col, from_json
  from pyspark.sql.types import StringType
  
  df = (spark.readStream   
      .format("kafka")   
      # ... 其他配置 ...
      .load()
      .select(col("value").cast(StringType()).alias("json"))
      # 启用PERMISSIVE模式捕获解析错误
      .select(
          from_json("json", jsonSchema, mode="PERMISSIVE").alias("data"),
          col("json").alias("raw_json")
      )
      .select("data.*", "raw_json"))
  

  之后可在DLT表中查看raw_json字段，确认是否存在解析失败的记录。

5. 查看Pipeline日志 #

• 直接在Databricks Pipeline页面查看运行日志，重点关注Driver Logs或Executor Logs中的错误信息，比如连接超时、权限拒绝、Schema不匹配等，这些日志是定位问题的关键。

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