Spark如何读取Kafka/Azure IoT Hub中的多记录Avro消息

问题根因 #

Spark内置的from_avro函数仅支持解析单条Avro记录的二进制编码，无法识别Avro容器格式（Object Container File）。通过C#AvroContainer写入的单条Kafka消息，实际是一个完整的微型Avro文件：自带格式头、Deflate压缩标记、同步标识，内部打包了多条同结构记录，因此原生方法只能解析到第一条记录，甚至直接返回解析失败。

实现步骤 #

前置依赖 #

在Databricks集群的PyPI库中安装fastavro包，用于直接解析Avro容器流，该库会自动识别压缩格式、容器内置schema，无需手动配置Deflate参数。

完整代码 #

from pyspark.sql.functions import udf, explode, col
from pyspark.sql.types import ArrayType, StructType, StructField, StringType, FloatType
import fastavro
from io import BytesIO

# 定义和业务Avro结构匹配的Spark Schema
sensor_schema = StructType([
    StructField("Timestamp", StringType(), False),
    StructField("SensorId", StringType(), False),
    StructField("SensorValue", FloatType(), True)
])

# 定义UDF，输入为Kafka消息的二进制value，输出为多条记录组成的数组
@udf(returnType=ArrayType(sensor_schema))
def parse_avro_container(value_bin):
    if not value_bin:
        return []
    with BytesIO(value_bin) as stream:
        # 自动识别容器格式、压缩编码、内置写入schema，迭代读取所有记录
        records = list(fastavro.reader(stream))
    # 做简单类型兼容，避免Spark类型校验报错
    return [
        {
            "Timestamp": str(r["Timestamp"]),
            "SensorId": str(r["SensorId"]),
            "SensorValue": float(r["SensorValue"]) if r["SensorValue"] is not None else None
        } for r in records
    ]

# 原有Kafka连接配置无需修改
kafka_df = spark.read\
            .format("kafka")\
            .option("subscribe", "topic-name")\
            .option("kafka.bootstrap.servers", "my-iothub-workspace.servicebus.windows.net:9093") \
            .option("kafka.security.protocol", "SASL_SSL")\
            .option("kafka.sasl.mechanism", "PLAIN")\
            .option("kafka.sasl.jaas.config", f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{connection_string}";')\
            .load()

# 解析多记录Avro，炸开数组得到单条明细记录
result_df = kafka_df.select(
        parse_avro_container(col("value")).alias("records")
    ).select(
        explode(col("records")).alias("data")
    ).select("data.*")

display(result_df)

优化选项 #

如果数据吞吐量极大，对解析延迟要求高，可以把上述UDF逻辑用Scala重写，直接调用Apache Avro官方Java库实现容器解析，避免Python进程序列化开销，逻辑和上述Python版本完全一致：将二进制value包装为字节流，创建DataFileReader迭代读取所有记录，返回数组后用explode拆分为单条即可。

注意事项 #

• 无需手动传入Avro schema给解析器：Avro容器头已经嵌入了写入时的完整schema，解析器会自动适配；如果写入时关闭了schema嵌入，再手动传入自定义schema即可。
• 无需手动配置Deflate压缩：Avro容器的元数据中已经标记了压缩算法，解析库会自动解码。
• 不要尝试调整原生from_avro的参数适配多记录场景，该函数从设计上就不支持Avro容器格式，只支持单条记录的裸二进制编码。

内容的提问来源于stack exchange，提问作者Daniel Argüelles