在Google Colab中使用PySpark读取Kafka流数据无输出问题求助

调整消费起始偏移量 #

默认情况下，Structured Streaming会从Kafka主题的最新偏移量开始消费。如果你的CSV数据是提前导入的，必须指定从最早偏移量读取，才能获取已存在的数据：

df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "your_topic_name") \
    .option("startingOffsets", "earliest") \  # 核心配置
    .load()

正确解析Kafka二进制消息 #

Kafka返回的value字段是二进制格式，需要先转为字符串，再按CSV结构解析：

from pyspark.sql.functions import col, from_csv
from pyspark.sql.types import StructType, StringType, IntegerType

# 定义与你的CSV结构匹配的Schema
csv_schema = StructType() \
    .add("column1", StringType()) \
    .add("column2", IntegerType()) \
    .add("column3", StringType())

# 解析消息内容
parsed_df = df.select(
    col("value").cast(StringType()),
    from_csv(col("value").cast(StringType()), csv_schema).alias("parsed_data")
).select("parsed_data.*")

配置流查询的输出规则 #

使用正确的输出模式并保持流运行，同时关闭控制台内容截断：

query = parsed_df.writeStream \
    .outputMode("append") \  # 新增数据场景用append，聚合场景用complete
    .format("console") \
    .option("truncate", "false") \
    .start()

query.awaitTermination()

检查版本兼容性 #

确保Spark的Kafka客户端包版本与你搭建的Kafka版本匹配，比如Spark 3.3.x对应spark-sql-kafka-0-10_2.12:3.3.0，启动SparkSession时指定：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("KafkaStreamReader") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0") \
    .getOrCreate()

验证Kafka主题数据可用性 #

在Colab终端用Kafka命令行工具确认主题有数据：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic your_topic_name --from-beginning

如果这里能输出数据，说明问题在Spark流配置；如果无输出，先排查CSV数据是否成功导入Kafka主题。

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