如何高效提取PySpark DataFrame JSON列的RemainingTime值转为Pandas DF

原代码性能差根因 #

原实现中每次循环都调用collect()拉取全量DataFrame数据到Driver节点，数据量大时会重复执行Spark作业多次，不仅运行速度极慢，还容易触发Driver内存溢出问题。

最优实现方案 #

该方案所有解析、提取逻辑都在Spark集群分布式执行，仅在最终需要转Pandas时拉取一次结果，性能远高于循环写法。

步骤1：导入依赖并定义JSON结构schema #

根据你给出的JSON样例，预先定义对应结构即可，不需要修改原表结构：

from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, ArrayType, BooleanType
import pandas as pd

# 匹配JSON结构的schema
json_schema = StructType([
    StructField("RXBatteryCharge", StringType(), nullable=True),
    StructField("RXBatteryLoad", StringType(), nullable=True),
    StructField("RXBatteryTime", StringType(), nullable=True),
    StructField("RemainingTime", ArrayType(IntegerType()), nullable=True),
    StructField("RemoteEnable", ArrayType(IntegerType()), nullable=True),
    StructField("SelectedMap", StringType(), nullable=True),
    StructField("SignalDoor", BooleanType(), nullable=True),
    StructField("SignalFailure", BooleanType(), nullable=True),
    StructField("SignalInfo", BooleanType(), nullable=True),
    StructField("SpinningSpeed", StringType(), nullable=True),
    StructField("StandbyState", StringType(), nullable=True),
    StructField("StartTime", ArrayType(IntegerType()), nullable=True),
    StructField("Status", IntegerType(), nullable=True),
    StructField("TargetTemperature", ArrayType(IntegerType()), nullable=True)
])

步骤2：提取目标字段 #

分两种场景适配：

• 如果你的json列是字符串类型，需要先解析再提取：

spark_result = df.withColumn("parsed_json", F.from_json(F.col("json"), json_schema)) \
                 .select(F.col("parsed_json.RemainingTime").alias("RemainingTime"))

• 如果你的json列已经是结构化Struct类型（读入时已自动解析），直接提取即可：

spark_result = df.select(F.col("json.RemainingTime").alias("RemainingTime"))

步骤3：转为Pandas DataFrame #

如果需要把数组类型的RemainingTime拆分为两个独立列，可以加一步拆分逻辑：

# 可选：拆分RemainingTime数组为两列
spark_result = spark_result.select(
    F.col("RemainingTime")[0].alias("RemainingTime_1"),
    F.col("RemainingTime")[1].alias("RemainingTime_2")
)

# 一次性转Pandas
pd_df = spark_result.toPandas()

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