在Databricks（PySpark）中从Kafka流数据时如何即时打印消费消息？

方法1：用Databricks原生display()实时展示结构化数据 #

从Kafka读取的原始数据是二进制的key和value，先解析为可读格式后直接调用display()，就能得到实时更新的无限表格：

data = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "你的Bootstrap服务器地址")
    .option("subscribe", "目标Kafka主题名")
    .load()
    # 将二进制value解析为字符串，也可根据实际格式解析为JSON/结构化数据
    .selectExpr("CAST(value AS STRING) AS message")
)

# 直接调用display，默认以流模式实时刷新展示
display(data)

运行后在Databricks可视化面板中，默认就是「Stream」模式，新消息进来会自动追加到表格里。

方法2：直接在控制台/日志打印每条消息 #

如果需要把消息逐条打印出来，可用foreachBatch实现轻量自定义逻辑，无需写入内存表：

# 定义批量打印逻辑
def print_messages(df, batch_id):
    # 遍历当前批次所有行并打印
    for row in df.collect():
        print(f"收到消息: {row.message}")

# 启动流并执行打印逻辑
data.writeStream.foreachBatch(print_messages).start().awaitTermination()

打印内容会输出到Databricks作业的「Logs」标签页，本地调试时直接输出到控制台。

注意事项 #

• 如果display()未自动更新流数据，可显式设置连续触发模式：

  data = data.trigger(processingTime='0 seconds')
  display(data)
  

• 若Kafka消息是JSON格式，可进一步解析为结构化DataFrame后展示：

  from pyspark.sql.functions import from_json, col
  from pyspark.sql.types import StructType, StringType
  
  schema = StructType().add("field1", StringType()).add("field2", StringType())
  data = data.select(from_json(col("message"), schema).alias("data")).select("data.*")
  display(data)
  

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