PySpark读取Kafka流无输出，控制台空白求助

核心问题定位 #

你的代码存在关键时序问题：批处理写入Kafka的操作早于流读取启动。批处理是一次性完成写入，而默认情况下Spark流读取会从Kafka的最新偏移量开始消费，导致启动时已经错过之前写入的所有消息，因此控制台无输出。

具体修复步骤 #

1. 调整执行顺序：先启动流读取，再写入数据 #

将批处理写入代码放到流查询启动之后，确保流读取已经在监听topic，能实时捕获后续写入的消息：

from pyspark.sql import SparkSession
from pyspark.sql.functions import expr

# 创建Spark会话
spark = SparkSession.builder \
    .appName("KafkaStreamWriter") \
    .config('spark.jars.packages', 'org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.0') \
    .getOrCreate()

kafka_bootstrap_servers = "localhost:29092"

# 先启动流读取任务
df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", "taxi_topic") \
    .load()

# 转换Kafka消息格式
df = df.selectExpr("CAST(value AS STRING) AS value").select("value")

# 启动控制台输出流
query = df \
    .writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

# 执行批处理写入Kafka
data = [(i,) for i in range(1, 101)]
batch_df = spark.createDataFrame(data, ["number"])

batch_df.selectExpr("CAST(number AS STRING) AS value") \
    .write \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("topic", "taxi_topic") \
    .save()

# 等待流任务结束
query.awaitTermination()

2. 读取历史消息：指定从最早偏移量开始消费 #

如果需要读取topic中已存在的历史数据，给readStream添加startingOffsets参数：

df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", "taxi_topic") \
    .option("startingOffsets", "earliest")  # 从topic最早的偏移量开始消费
    .load()

3. 验证网络连通性 #

虽然终端消费者正常，但需确认Spark运行环境能访问Kafka的29092端口：

• 若Spark本地运行，检查Docker端口映射是否正确（29092已暴露到宿主机）
• 若Spark也在Docker容器中，需使用Kafka容器的内部地址或Docker网络别名，而非localhost

4. 检查Kafka Topic状态 #

用Kafka命令行工具确认消息是否成功写入topic：

# 查看topic分区与偏移量信息
kafka-topics.sh --describe --topic taxi_topic --bootstrap-server localhost:29092

额外注意事项 #

• 确保Spark版本与Kafka连接器版本匹配，你使用的spark-sql-kafka-0-10_2.12:3.4.0对应Spark 3.4.x，版本需一致
• 流查询启动后为异步运行，必须保证写入操作在流启动之后执行，否则会错过消息

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