Databricks中PySpark解析EventHub解码字符串为JSON的方案咨询

EventHub生成的Avro文件落地后，经AutoLoader加载入表，Body字段解码为字符串列decoded。该列每行格式为：

• 首行是逗号分隔的键列表
• 次行是逗号分隔的对应值列表
  需要将每行批量转换为JSON格式存储，现有代码仅能处理单行，寻求PySpark高效批量实现方案。

输入示例：

@300://mm/cm#//c.Process/p.TriggerAck/v,@2222,@300://imm/cm#//c.AcousticAlarm1/p.sv_bAlarm/v,@300://imm/cm#//c.AirValve/p.sv_bActivatedInSequence/v
444444,4.73,0

期望输出JSON：

{
  "@300://mm/cm#//c.Process/p.TriggerAck/v": 444444,
  "@2222": 4.73,
  "@300://imm/cm#//c.AirValve/p.sv_bActivatedInSequence/v": 0
}

现有单行处理代码：

import pandas as pd
from io import StringIO

data = spark.sql("SELECT decoded FROM catalog.schema.table")

first_row = data.head()

decoded_body = first_row[0]

df = pd.read_csv(StringIO(decoded_body))

用PySpark内置分布式函数实现批量处理，充分利用Spark的分布式计算能力，避免低效的单行迭代：

步骤1：拆分键行与值行 #

将decoded字符串按换行符分割，拆分出单独的键行和值行：

from pyspark.sql import functions as F

# 读取源表数据
df = spark.sql("SELECT decoded FROM catalog.schema.table")

# 按换行分割为行数组，提取首行（键）和次行（值）
split_lines_df = df.withColumn("lines", F.split(F.col("decoded"), "\n")) \
                   .withColumn("keys_str", F.element_at(F.col("lines"), 1)) \
                   .withColumn("values_str", F.element_at(F.col("lines"), 2))

步骤2：拆分键值为数组 #

将键行和值行按逗号分割成数组，为后续配对做准备：

split_arrays_df = split_lines_df.withColumn("keys", F.split(F.col("keys_str"), ",")) \
                               .withColumn("values", F.split(F.col("values_str"), ","))

步骤3：键值配对并转换为JSON #

用map_from_arrays将键数组与值数组配对成Map类型，再转换为JSON字符串；同时自动转换值的类型（整数/浮点数）：

# 定义UDF自动识别值类型
def convert_value(value_str):
    try:
        return int(value_str)
    except ValueError:
        try:
            return float(value_str)
        except ValueError:
            return value_str

convert_value_udf = F.udf(convert_value)

# 转换值类型、生成键值Map、转JSON
result_df = split_arrays_df.withColumn("values_casted", F.transform(F.col("values"), convert_value_udf)) \
                           .withColumn("key_value_map", F.map_from_arrays(F.col("keys"), F.col("values_casted"))) \
                           .withColumn("json_output", F.to_json(F.col("key_value_map"))) \
                           .select("json_output")

步骤4：存储结果 #

将转换后的JSON列写入目标表：

result_df.write.mode("append").saveAsTable("catalog.schema.target_table")

无UDF优化方案（值类型统一场景） #

如果值类型固定或不需要自动识别，可直接用内置函数生成Map转JSON，性能更优：

result_df = split_arrays_df.withColumn("key_value_map", F.map_from_arrays(F.col("keys"), F.col("values"))) \
                           .withColumn("json_output", F.to_json(F.col("key_value_map"))) \
                           .select("json_output")

注意事项 #

• 若存在键数与值数不匹配的异常行，可添加过滤条件：F.size(F.col("keys")) == F.size(F.col("values"))
• map_from_arrays需Spark 2.4及以上版本支持

内容的提问来源于stack exchange，提问作者Yami Mahō