如何将Databricks Delta Live Tables流数据写入Kafka实例？

1. 先明确Kafka Sink的硬性要求 #

Spark往Kafka写流时，对DataFrame的结构有明确规则：

• 必须包含key和value两个字段，类型均为String（二进制类型也支持，但新手用String更省心）
• 可选字段：若要动态指定写入topic，可添加topic字段；指定分区加partition；指定时间戳加timestamp

2. 将实时表转换为符合要求的格式 #

假设你的实时表名为real_time_table，先取出它的数据流，再做格式转换：

场景1：整行转JSON作为value，用表中字段当key #

比如你的表有id、name、update_time字段，把整行数据打包成JSON字符串作为value，用id作为key：

import org.apache.spark.sql.functions._

// 获取实时表的数据流
val realTimeStream = spark.readStream.table("real_time_table")

// 转换为Kafka可接受的格式
val kafkaReadyStream = realTimeStream
  .select(
    col("id").cast("string").alias("key"),  // 将id转为String类型作为key
    to_json(struct("*")).alias("value")     // 把所有字段转成JSON字符串作为value
  )

场景2：只选部分字段生成value #

如果不需要全量字段，仅挑选特定字段转成JSON：

val kafkaReadyStream = realTimeStream
  .select(
    col("id").cast("string").alias("key"),
    // 仅将name和update_time转成JSON
    to_json(struct("name", "update_time")).alias("value")
  )

场景3：用固定字符串当key #

如果不需要自定义key，直接使用固定值：

val kafkaReadyStream = realTimeStream
  .select(
    lit("default_key").alias("key"),  // 固定key值
    to_json(struct("*")).alias("value")
  )

3. 写入Kafka实例 #

基于你提供的代码，结合转换后的数据流完成写入，注意必须配置checkpoint路径：

kafkaReadyStream
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "updates")  // 全局指定写入的topic，若数据流中包含topic字段会覆盖此配置
  .option("checkpointLocation", "/path/to/your/checkpoint")  // 必须设置，用于记录流处理进度，避免重复消费
  .start()
  .awaitTermination()  // 保持程序运行，持续处理数据流

必看注意事项 #

• Checkpoint必须配置：生产环境建议用分布式存储路径（如HDFS），本地路径仅适合测试；它能保证程序重启后接续上次进度处理，不会重复处理数据
• 字段类型不能错：key和value必须是String或Binary类型，其他类型会直接报错，记得用cast("string")转换
• 先调试再写入：不确定格式是否正确时，可先将流输出到控制台验证：kafkaReadyStream.writeStream.format("console").start().awaitTermination()，确认key和value内容无误后再写入Kafka
• 优先用JSON格式：将结构化数据转成JSON，消费Kafka时可方便解析回原结构，是新手最通用的选择

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