使用PySpark ReadStream从Kafka读取Avro数组记录遇到问题

正确的Avro数组Schema定义 #

直接在原单条记录Schema外层定义数组类型即可，无需简单拼接[]，正确的Schema结构如下：

{
    "type": "array",
    "items": {
        "type": "record",
        "name": "data",
        "fields": [
            {
                "name": "x",
                "type": ["double", "null"]
            },
            {
                "name": "y",
                "type": ["double", "null"]
            }
        ]
    }
}

解析步骤与代码示例 #

使用Spark原生from_avro函数结合上述Schema完成解析，按需可将数组展开为单独记录行：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, explode
from pyspark.sql.avro.functions import from_avro

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("KafkaAvroArrayParser") \
    .getOrCreate()

# 定义数组类型的Avro Schema
avro_array_schema = """
{
    "type": "array",
    "items": {
        "type": "record",
        "name": "data",
        "fields": [
            {
                "name": "x",
                "type": ["double", "null"]
            },
            {
                "name": "y",
                "type": ["double", "null"]
            }
        ]
    }
}
"""

# 读取Kafka流
kafka_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-bootstrap-server:9092") \
    .option("subscribe", "target-topic") \
    .load()

# 解析Avro格式的数组字段
parsed_stream = kafka_stream.select(
    # 可选：保留Kafka元数据
    col("key").cast("string").alias("kafka_key"),
    col("timestamp").alias("kafka_timestamp"),
    # 解析value为Avro数组
    from_avro(col("value"), avro_array_schema).alias("data_array")
)

# 可选：将数组展开为单条记录行
exploded_stream = parsed_stream.select(
    "kafka_key",
    "kafka_timestamp",
    explode(col("data_array")).alias("single_data")
).select(
    "kafka_key",
    "kafka_timestamp",
    "single_data.x",
    "single_data.y"
)

# 输出到控制台（测试用）
query = exploded_stream.writeStream \
    .outputMode("append") \
    .format("console") \
    .option("truncate", "false") \
    .start()

query.awaitTermination()

注意事项 #

• 确认Kafka主题中的消息是Avro编码的数组，而非JSON格式数组（若为JSON需改用from_json函数）
• Schema中的name字段需保证唯一性，若有同名Schema可添加namespace字段区分
• 若不需要展开数组，可直接对data_array字段进行过滤、聚合等后续操作

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