如何在PySpark中解析Protobuf生成的规范JSON格式数据？

针对你遇到的Protobuf规范JSON与.proto字段命名不匹配的问题，这里提供两种实用的解决方法：

方法一：利用Protobuf Python库结合Spark UDF #

Protobuf官方库本身支持将规范JSON解析为对应的消息对象，我们可以通过Spark UDF把这个能力整合到PySpark中：

1. 编译.proto文件为Python类
   用protoc编译器生成Python版的消息类：

   protoc --python_out=. my_message.proto
   

   执行后会生成my_package_pb2.py文件。

2. 编写Spark UDF解析JSON
   在PySpark中导入生成的类，编写UDF将JSON字符串解析为符合.proto字段名的Row：

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import udf
   from pyspark.sql.types import StructType, StructField, StringType
   import my_package_pb2
   
   # 定义输出Schema，与.proto字段一致
   output_schema = StructType([
       StructField("my_field", StringType(), nullable=True),
       StructField("my_long_name_field", StringType(), nullable=True)
   ])
   
   @udf(returnType=output_schema)
   def parse_protobuf_json(json_str):
       msg = my_package_pb2.MyMessage()
       msg.ParseJson(json_str)
       return (msg.my_field, msg.my_long_name_field)
   
   # 初始化SparkSession并处理数据
   spark = SparkSession.builder.appName("ProtobufJsonParser").getOrCreate()
   df = spark.read.text("path/to/json/files")  # 读取原始JSON文本
   parsed_df = df.withColumn("parsed_data", parse_protobuf_json("value")).select("parsed_data.*")
   parsed_df.show()
   

   这种方法完全遵循Protobuf的JSON映射规则，无需手动维护字段映射，适合复杂的.proto结构。

方法二：手动构建字段映射字典转换列 #

如果你的.proto结构简单，也可以手动构建JSON键到.proto字段名的映射，直接在Spark中重命名列：

1. 定义字段映射
   根据你的.proto和JSON格式，创建映射字典：

   field_mapping = {
       "myField": "my_field",
       "short": "my_long_name_field"
   }
   

2. 在Spark中转换列
   先解析JSON为StructType，再通过selectExpr或withColumn重命名列：

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import from_json
   from pyspark.sql.types import StructType, StructField, StringType
   
   spark = SparkSession.builder.appName("ProtobufJsonMapper").getOrCreate()
   
   # 定义输入JSON的Schema
   input_schema = StructType([
       StructField("myField", StringType(), nullable=True),
       StructField("short", StringType(), nullable=True)
   ])
   
   # 读取JSON数据并解析
   df = spark.read.json("path/to/json/files", schema=input_schema)
   # 生成重命名表达式
   rename_exprs = [f"`{json_key}` AS {proto_field}" for json_key, proto_field in field_mapping.items()]
   parsed_df = df.selectExpr(*rename_exprs)
   parsed_df.show()
   

   这种方法无需依赖Protobuf编译后的类，适合结构固定且简单的场景，但需要手动维护映射关系，结构变更时需同步更新。

注意事项 #

• 如果使用方法一，确保PySpark集群的所有节点都能访问到编译后的Protobuf Python类文件。
• 对于嵌套的Protobuf消息结构，方法一的扩展性更好，Protobuf库会自动处理嵌套字段的映射。

内容的提问来源于stack exchange，提问作者Vito De Tullio