如何优化PySpark解析大型嵌套JSON文件的处理速度？

你的代码和配置存在多处可优化点，调整后处理速度可提升数倍：

一、代码逻辑重构（核心优化点） #

原有代码的三个核心性能瓶颈：

• 本地读取整个400MB JSON计算result_count，相当于文件被重复读取2次，且大文件拉取到Driver端会产生极高内存开销
• 读取JSON时未传入预定义的schema，Spark自动推断schema需要全量扫描文件，浪费大量时间
• 转换逻辑冗余，多次循环加列、手动生成索引再explode的操作引入了大量不必要的计算开销

优化后代码如下：

import json
import pyspark.sql.functions as F
from pyspark.sql.types import StructType
from util import schema, meta_date

# 复用已有schema，省略Spark自动推断schema的扫描开销
new_schema = StructType.fromJson(json.loads(schema))

# 直接在构建Session时传入配置，不用启停两次上下文
spark = SparkSession.builder.master("spark://IP") \
    .config('spark.executor.memory', '5g') \
    .config('spark.executor.cores', '4') \
    .config('spark.driver.memory', '4g') \
    .config('spark.sql.shuffle.partitions', 8) \
    .config('spark.executor.extraJavaOptions', '-XX:+UseG1GC') \
    .getOrCreate()

# 读取时直接传入预定义schema
df = spark.read.schema(new_schema).json("largefile.json")

# 一步炸开result数组，无需手动计算索引
df = df.select(F.explode("data.result").alias("result"))
# 直接展开metric下所有维度字段，同时保留values数组
df = df.select("result.values", "result.metric.*")
# 炸开values数组拆分时间、数值字段
df = df.select(F.explode("values").alias("value_arr"), *meta_date) \
    .withColumn("time", F.col("value_arr").getItem(0)) \
    .withColumn("value", F.col("value_arr").getItem(1)) \
    .drop("value_arr")

df.show()

二、集群配置&存储优化 #

• 原有内存、核数配置基本合理，新增的spark.sql.shuffle.partitions=8适配你的总逻辑核数，避免默认200分区带来的过度调度开销；新增G1GC配置优化JSON解析场景下的大量临时对象回收效率
• 单文件JSON无法发挥并行读取优势，可提前将400MB的JSON拆分为4~8个100MB以内的小文件，能让多个CPU核心并行处理，进一步缩短读取时间

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