打印并验证 PySpark Structured Streaming 读取的 Kafka 主题数据

你的 Structured Streaming 代码没有输出，主要有以下几个原因：

• spark.readStream.format("kafka").load() 只是定义流式读取逻辑，不会立即读取数据。
• spark.sql("select * from kafka") 也只是生成一个新的流式 DataFrame，不会触发执行。
• Structured Streaming 必须通过 writeStream.start() 启动查询。
• Kafka 的 value 字段默认是二进制类型，需要先使用 CAST(value AS STRING) 转成字符串，再用 from_json 解析 JSON。
• 你的 Producer 中 p1.send() 在循环外部，因此实际只发送了最后一个文件的内容。
• 示例 JSON 最外层使用了 {}，但里面是多条记录且没有字段名，这不是合法 JSON。多条记录应使用数组 []。
• 不需要同时使用 KafkaConsumer 和 PySpark Structured Streaming。使用 Structured Streaming 时，Spark 会自己从 Kafka 读取数据。

多条记录应写成 JSON 数组：

[
  {
    "Customer Number": "1",
    "Customer Name": "Aditya",
    "Restaurant Number": "2201",
    "Restaurant NameOrdered": "Bawarchi",
    "Number of Items": "3",
    "price": "10",
    "Operating Start hours": "9:00",
    "Operating End hours": "23:00"
  },
  {
    "Customer Number": "2",
    "Customer Name": "Sarva",
    "Restaurant Number": "2202",
    "Restaurant NameOrdered": "Sarvana Bhavan",
    "Number of Items": "4",
    "price": "20",
    "Operating Start hours": "8:00",
    "Operating End hours": "20:00"
  },
  {
    "Customer Number": "3",
    "Customer Name": "Kala",
    "Restaurant Number": "2203",
    "Restaurant NameOrdered": "Taco Bell",
    "Number of Items": "5",
    "price": "30",
    "Operating Start hours": "11:00",
    "Operating End hours": "21:00"
  }
]

建议先把 S3 文件内容解析成 Python 对象，再把每条记录分别发送到 Kafka。这样每条 Kafka 消息就是一条 JSON 记录，Spark 解析时最简单。

import json
import boto3
from kafka import KafkaProducer


class Producer:
    @staticmethod
    def json_serializer(data):
        return json.dumps(data).encode("utf-8")

    @staticmethod
    def read_s3():
        producer = KafkaProducer(
            bootstrap_servers=["localhost:9092"],
            value_serializer=Producer.json_serializer
        )

        s3 = boto3.resource("s3")
        bucket = s3.Bucket("kakfa")

        for obj in bucket.objects.all():
            body = obj.get()["Body"].read().decode("utf-8")
            records = json.loads(body)

            # 如果文件里是单条对象，就包装成列表
            if isinstance(records, dict):
                records = [records]

            # 每条记录单独发送一条 Kafka 消息
            for record in records:
                producer.send("Uber_Eats", record)

        producer.flush()
        producer.close()

注意：原来的代码先把文件内容解码成字符串，再交给 json.dumps()，会导致 JSON 被重复序列化。上面的代码先使用 json.loads() 解析，再发送字典对象。

下面代码先读取 Kafka，再把 value 转成字符串，最后根据定义好的 Schema 解析 JSON，并输出到控制台。

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json
from pyspark.sql.types import StructType, StructField, StringType


class Consumer:
    @staticmethod
    def read_from_topic(spark):
        schema = StructType([
            StructField("Customer Number", StringType(), True),
            StructField("Customer Name", StringType(), True),
            StructField("Restaurant Number", StringType(), True),
            StructField("Restaurant NameOrdered", StringType(), True),
            StructField("Number of Items", StringType(), True),
            StructField("price", StringType(), True),
            StructField("Operating Start hours", StringType(), True),
            StructField("Operating End hours", StringType(), True)
        ])

        kafka_df = (
            spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9092")
            .option("subscribe", "Uber_Eats")
            .option("startingOffsets", "earliest")
            .load()
        )

        # Kafka 的 value 是二进制，先转成字符串
        json_df = kafka_df.selectExpr(
            "CAST(value AS STRING) AS json_string"
        )

        # 解析 JSON，并展开每个字段
        parsed_df = (
            json_df
            .select(from_json(col("json_string"), schema).alias("data"))
            .select("data.*")
        )

        # 启动流式查询，输出到控制台
        query = (
            parsed_df.writeStream
            .format("console")
            .option("truncate", False)
            .start()
        )

        # 等待查询持续运行
        query.awaitTermination()

如果只使用 Structured Streaming，就不需要再调用 KafkaConsumer。

from Producer_Group import Producer
from Consumer_Group import Consumer
from Spark_Connection import connection


class Client:
    @staticmethod
    def transfer():
        spark = connection.get_connection()

        Producer.read_s3()
        Consumer.read_from_topic(spark)


client = Client()
client.transfer()

read_from_topic() 中的 awaitTermination() 会阻塞主程序，使流式查询一直运行。如果只是本地短时间测试，也可以给它设置超时时间，例如 awaitTermination(60000)。

数据能正常解析后，可以在写入数据库之前先做条件校验。例如筛选价格大于 10 的记录：

from pyspark.sql.functions import col

checked_df = parsed_df.filter(col("price").cast("int") > 10)

query = (
    checked_df.writeStream
    .format("console")
    .option("truncate", False)
    .start()
)

query.awaitTermination()

控制台输出确认无误后，再把 console sink 换成数据库对应的 sink。

如果你坚持把整个 JSON 数组作为一条 Kafka 消息发送，那么 Spark 端需要先解析数组，再使用 explode 展开：

from pyspark.sql.types import ArrayType
from pyspark.sql.functions import explode

array_schema = ArrayType(schema)

parsed_df = (
    json_df
    .select(from_json(col("json_string"), array_schema).alias("records"))
    .select(explode(col("records")).alias("data"))
    .select("data.*")
)

不过更推荐第一种方式：Producer 中每条记录发送一条 Kafka 消息。

本地运行时，需要确保 Spark 包含 Kafka connector。可以在 spark-submit 中添加对应包，Scala 和 Spark 版本要与本地环境一致：

spark-submit \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:<你的Spark版本> \
  your_script.py

核心点是：流式 DataFrame 必须调用 writeStream.start()，并且 Kafka 的 value 必须先转成 STRING，再用 from_json 按 Schema 解析。

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