Spark中Avro文件反序列化后值为null的问题求助

核心问题原因 #

你当前的流程存在关键错误：从Kafka读取的value本身就是Avro序列化后的二进制数据，但你直接将这个二进制字段以Avro格式写入HDFS时，Spark会把DataFrame的结构（仅一个binary类型的value字段）封装成Avro文件。最终生成的Avro文件的Schema是包含value: binary的结构，而非你定义的GPRSIOT Schema。当你用GPRSIOT Schema读取时，字段完全不匹配，导致大部分字段返回Null（只有设置了默认值的字段能显示默认值）。

修复方案 #

方案1：修正数据写入流程（推荐） #

直接在读取Kafka数据时完成Avro反序列化，再将结构化数据写入Avro文件：

# 读取Kafka并反序列化Avro数据
df = (
    spark.read.format("kafka")
    .option("kafka.security.protocol", "PLAINTEXT")
    .option("kafka.sasl.mechanism", "GSSAPI")
    .option("kafka.bootstrap.servers", KAFKA_BROKERS)
    .option("subscribe", KAKFA_TOPIC)
    .option("group.id", "noid")
    .option("startingOffsets", "earliest")
    .option("failOnDataLoss", "false")
    .option("maxOffsetsPerTrigger", 2)
    .load()
    # 使用deserialize_avro函数解析Kafka的value字段
    .selectExpr(f"deserialize_avro(value, '{schema}') as data")
    # 展开结构化数据的所有字段
    .select("data.*")
)

# 将结构化数据写入Avro文件
df.write.format("avro").save("path/to/correct_avro_files")

方案2：修复已生成的错误Avro文件 #

如果已经有了错误生成的Avro文件，可以读取其中的value二进制字段，再重新反序列化：

# 读取错误的Avro文件（此时Schema是包含value: binary的结构）
wrong_df = spark.read.format("avro").load("path/to/avro_files")

# 解析二进制value为Avro结构化数据
correct_df = wrong_df.selectExpr(f"deserialize_avro(value, '{schema}') as data").select("data.*")

# 查看正确数据
correct_df.show()

注意事项 #

• 确保你的Spark环境已安装spark-avro依赖包，版本需与Spark版本兼容（例如Spark 3.3.x对应org.apache.spark:spark-avro_2.12:3.3.0）。
• 验证你提供的GPRSIOT Schema与Kafka中原始Avro数据的Schema完全一致，字段名、类型、顺序必须匹配，否则仍会出现Null值。

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