Pyspark 解析高度嵌套的 json (Prometheus)

Posted 2023-04-15

【中文标题】Pyspark 解析高度嵌套的 json (Prometheus)

【英文标题】：Pyspark parse highly nested json (Prometheus)

【发布时间】：2021-11-21 13:15:40

【问题描述】：

我真的很想在使用 PySpark-SQL 解析嵌套 JSON 数据方面得到一些帮助，因为我是 PySpark 的新手。数据具有以下架构：

架构

root

|-- data: struct (nullable = true)

|    |-- result: array (nullable = true)

|    |    |-- element: struct (containsNull = true)

|    |    |    |-- metric: struct (nullable = true)

|    |    |    |    |-- data0: string (nullable = true)

|    |    |    |    |-- data1: string (nullable = true)

|    |    |    |    |-- data2: string (nullable = true)

|    |    |    |    |-- data3: string (nullable = true)

|    |    |    |-- values: array (nullable = true)

|    |    |    |    |-- element: array (containsNull = true)

|    |    |    |    |    |-- element: string (containsNull = true)

|    |-- resultType: string (nullable = true)

|-- status: string (nullable = true)

这是 JSON 文件（输入）的示例：

“状态”：“成功”，

“数据”：“结果类型”：“矩阵”，“结果”：

["metric":"data0":"T" ,"data1":"O","values":[[90,"0"],[80, "0"]],

"metric":"data0":"K" ,"data1":"S","values":[[70,"0"],[60, "0"]],

"metric":"data2":"J" ,"data3":"O","values":[[50,"0"],[40, "0"]]]

我的目标我基本上希望将数据放入以下数据框中：

1-

data0 | data1 | data2 | data3 |values

示例输出数据框：

data0  | data1 | data2  | data3 | values
"T"    |   "O" |    nan |    nan|   [90,"0"],[80, "0"]
"K"    |   "S" |    nan |    nan|   [70,"0"],[60, "0"]
nan    |   nan |    "J" |    "O"|   [50,"0"],[40, "0"]

2-

time | value | data0 | data1 | data2 | data3

示例输出数据帧

time | value |data0 | data1 | data2  | data3 
90   |   "0" |   "T"|    "O"|   nan  | nan
80   |   "0" |   "T"|    "O"|   nan  | nan
70   |   "0" |   "K"|    "S"|   nan  | nan
60   |   "0" |   "K"|    "S"|   nan  | nan
50   |   "0" |   nan|    nan|   "J"  | "O"
40   |   "0" |   nan|    nan|   "J"  | "O"

另外，如果有任何方法可以使用 spark 的并行能力来加速这个过程，那将是很好的，因为解析的 json 文件以千兆字节为单位。

【问题讨论】：

请提供输入输出数据示例。

请再次检查问题，我做了更新。谢谢

【参考方案1】：

要获取第一个数据帧，您可以使用：

df = (
    df.withColumn("data0", F.expr("transform(data.result, x -> x.metric.data0)"))
    .withColumn("data1", F.expr("transform(data.result, x -> x.metric.data1)"))
    .withColumn("data2", F.expr("transform(data.result, x -> x.metric.data2)"))
    .withColumn("data3", F.expr("transform(data.result, x -> x.metric.data3)"))
    .withColumn("values", F.expr("transform(data.result, x -> x.values)"))
    .withColumn("items", F.array(F.lit(0), F.lit(1), F.lit(2)))
    .withColumn("items", F.explode(F.col("items")))
    .withColumn("data0", F.col("data0").getItem(F.col("items")))
    .withColumn("data1", F.col("data1").getItem(F.col("items")))
    .withColumn("data2", F.col("data2").getItem(F.col("items")))
    .withColumn("data3", F.col("data3").getItem(F.col("items")))
    .withColumn("values", F.col("values").getItem(F.col("items")))
    .drop("data", "status", "items")
)

结果：

root
 |-- data0: string (nullable = true)
 |-- data1: string (nullable = true)
 |-- data2: string (nullable = true)
 |-- data3: string (nullable = true)
 |-- values: array (nullable = true)
 |    |-- element: array (containsNull = true)
 |    |    |-- element: string (containsNull = true)

+-----+-----+-----+-----+------------------+
|data0|data1|data2|data3|values            |
+-----+-----+-----+-----+------------------+
|T    |O    |null |null |[[90, 0], [80, 0]]|
|K    |S    |null |null |[[70, 0], [60, 0]]|
|null |null |J    |O    |[[50, 0], [40, 0]]|
+-----+-----+-----+-----+------------------+

要获得第二个，它是相同的，但附加了 explode 值：

df = (
    df.withColumn("data0", F.expr("transform(data.result, x -> x.metric.data0)"))
    .withColumn("data1", F.expr("transform(data.result, x -> x.metric.data1)"))
    .withColumn("data2", F.expr("transform(data.result, x -> x.metric.data2)"))
    .withColumn("data3", F.expr("transform(data.result, x -> x.metric.data3)"))
    .withColumn("values", F.expr("transform(data.result, x -> x.values)"))
    .withColumn("items", F.array(F.lit(0), F.lit(1), F.lit(2)))
    .withColumn("items", F.explode(F.col("items")))
    .withColumn("data0", F.col("data0").getItem(F.col("items")))
    .withColumn("data1", F.col("data1").getItem(F.col("items")))
    .withColumn("data2", F.col("data2").getItem(F.col("items")))
    .withColumn("data3", F.col("data3").getItem(F.col("items")))
    .withColumn("values", F.col("values").getItem(F.col("items")))
    .withColumn("values", F.explode("values"))
    .withColumn("time", F.col("values").getItem(0))
    .withColumn("value", F.col("values").getItem(1))
    .drop("data", "status", "items", "values")
)

结果：

root
 |-- data0: string (nullable = true)
 |-- data1: string (nullable = true)
 |-- data2: string (nullable = true)
 |-- data3: string (nullable = true)
 |-- time: string (nullable = true)
 |-- value: string (nullable = true)

+-----+-----+-----+-----+----+-----+
|data0|data1|data2|data3|time|value|
+-----+-----+-----+-----+----+-----+
|T    |O    |null |null |90  |0    |
|T    |O    |null |null |80  |0    |
|K    |S    |null |null |70  |0    |
|K    |S    |null |null |60  |0    |
|null |null |J    |O    |50  |0    |
|null |null |J    |O    |40  |0    |
+-----+-----+-----+-----+----+-----+

---

更新：

自动化data 名称和结果数量的示例：

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
import json

data_names = []
number_of_results = 0
with open("test.json", "r") as f_in:
    raw_data = json.load(f_in)
    for item in raw_data["data"]["result"]:
        number_of_results += 1
        for key in item["metric"].keys():
            if key not in data_names:
                data_names.append(key)

spark = SparkSession.builder.getOrCreate()
df = spark.read.option("multiline", True).json("test.json")

for data_name in data_names:
    df = df.withColumn(
        data_name, F.expr(f"transform(data.result, x -> x.metric.data_name)")
    )

df = (
    df.withColumn("values", F.expr("transform(data.result, x -> x.values)"))
    .withColumn("items", F.array(*[F.lit(x) for x in range(0, number_of_results)]))
    .withColumn("items", F.explode(F.col("items")))
)

for data_name in data_names:
    df = df.withColumn(data_name, F.col(data_name).getItem(F.col("items")))

df = df.withColumn("values", F.col("values").getItem(F.col("items"))).drop(
    "data", "status", "items"
)

结果是第一个数据帧（同上）

【讨论】：

感谢您的回答。如果我不知道度量字典的键怎么办？例如，可能是（data4、data6 等）是否有解决此问题的方法？

@Samo 不客气。您是否有权在将 JSON 作为数据帧加载之前对其进行解析？

是的，我有访问权限。

@Samo 更新以提供自动化数据名称的示例，但这实际上取决于您的数据集。有时，如果不探索输入数据集，就无法实现自动化。

感谢您，给您带来的不便，我们深表歉意