PySpark-SQL解析Prometheus嵌套JSON及大文件处理优化咨询

前置准备 #

首先初始化SparkSession并提前定义数据Schema，跳过自动Schema推断的额外开销：

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, col
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, LongType
from pyspark import StorageLevel

# 初始化SparkSession
spark = SparkSession.builder.appName("PrometheusJsonParse").getOrCreate()

# 自定义Schema和你给出的结构完全匹配
schema = StructType([
    StructField("status", StringType(), nullable=True),
    StructField("data", StructType([
        StructField("resultType", StringType(), nullable=True),
        StructField("result", ArrayType(StructType([
            StructField("metric", StructType([
                StructField("data0", StringType(), nullable=True),
                StructField("data1", StringType(), nullable=True),
                StructField("data2", StringType(), nullable=True),
                StructField("data3", StringType(), nullable=True)
            ]), nullable=True),
            StructField("values", ArrayType(ArrayType(StringType())), nullable=True)
        ])), nullable=True)
    ]), nullable=True)
])

# 读取源JSON文件
raw_df = spark.read.schema(schema).json("你的JSON文件路径")

第一种DataFrame实现 #

直接展开data.result数组，提取metric字段和完整values数组即可：

df1 = raw_df.select(explode(col("data.result")).alias("result")) \
    .select(
        col("result.metric.data0"),
        col("result.metric.data1"),
        col("result.metric.data2"),
        col("result.metric.data3"),
        col("result.values")
    )

# 查看结果
df1.show(truncate=False)

输出和你要求的结构完全一致，缺失的metric字段自动为null。

第二种DataFrame实现 #

在第一种结构的基础上二次展开values数组，拆分时间和数值字段：

df2 = raw_df.select(explode(col("data.result")).alias("result")) \
    .select(
        col("result.metric.*"),
        explode(col("result.values")).alias("value_arr")
    ) \
    .select(
        col("value_arr")[0].cast(LongType()).alias("time"),
        col("value_arr")[1].alias("value"),
        col("data0"),
        col("data1"),
        col("data2"),
        col("data3")
    )

# 查看结果
df2.show(truncate=False)

GB级大文件性能优化方案 #

• 强制指定Schema：嵌套JSON自动推断需要扫描全量数据，提前定义Schema可减少30%以上的读取耗时，同时避免Schema推断错误。
• 调整分区大小：修改spark.sql.files.maxPartitionBytes参数（默认128MB），根据集群CPU核心数设置单分区大小为64MB~256MB，保证并行任务数和CPU核心数匹配，避免并行不足或者小任务过多。
• 缓存中间结果：如果两个DataFrame都需要使用，可将展开data.result后的中间结果执行persist(StorageLevel.MEMORY_AND_DISK)缓存，不需要重复读取解析源文件。
• 存储优化：结果落地优先选择Parquet列式存储，比JSON节省70%以上存储空间，后续查询效率提升数倍，可根据业务常用过滤字段做分区存储。
• 小文件合并：如果源数据是大量KB级小文件，可调大spark.sql.files.openCostInBytes参数，或者用coalesce/repartition合并小分区，减少IO开销。

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