Pyspark将复杂的json解析为行[关闭]

Question

我在一个文件中有以下 json：

{
    "code": 0,
    "msg": "OK",
    "sensors": {
        "258443": {
            "fieldname": "DeviceStatus2",
            "info": "Unlock Request",
            "id": 258443,
            "work": {
                "1623020526": {
                    "ts": 1623020526,
                    "te": 1623021124,
                    "lat": "-25.911997",
                    "lon": "28.169119"
                },
                "1623021393": {
                    "ts": 1623021393,
                    "te": 1623021453,
                    "lat": "-25.911997",
                    "lon": "28.169119"
                },
                "1623052154": {
                    "ts": 1623052154,
                    "te": 1623052783,
                    "lat": "-25.790348",
                    "lon": "28.308369"
                }
            }
        },
        "258441": {
            "fieldname": "Dis1",
            "info": "Door",
            "id": 258441,
            "work": {
                "1623021447": {
                    "ts": 1623021447,
                    "te": 1623035978,
                    "lat": "-25.911997",
                    "lon": "28.169119"
                },
                "1623052775": {
                    "ts": 1623052775,
                    "te": 1623058925,
                    "lat": "-25.790348",
                    "lon": "28.308369"
                }
            }
        },
        "258442": {
            "fieldname": "AnalogInput",
            "info": "Lock Fire",
            "id": 258442,
            "work": {
                "1623021453": {
                    "ts": 1623021453,
                    "te": 1623021633,
                    "lat": "-25.911997",
                    "lon": "28.169119"
                },
                "1623052783": {
                    "ts": 1623052783,
                    "te": 1623052962,
                    "lat": "-25.790348",
                    "lon": "28.308369"
                }
            }
        }
    }
}

我需要使用 pyspark 将其解析为以下行：

258443,DeviceStatus2,Unlock Request,258443,1623020526,1623021124,-25.911997,28.169119
258443,DeviceStatus2,Unlock Request,258443,1623021393,1623021453,-25.911997,28.169119
258443,DeviceStatus2,Unlock Request,258443,1623052154,1623052783,-25.790348,28.308369
258441,Dis1,Door,258441,1623021447,1623035978,-25.911997,-25.911997
258441,Dis1,Door,258441,1623052775,1623058925,-25.790348,28.308369
258442,AnalogInput,Lock Fire258442,1623021453,1623021633,-25.911997,28.169119
258442,AnalogInput,Lock Fire258442,1623052783,1623052962,-25.790348,28.308369

我曾尝试使用explode 和arrays_zip，但失败了。 如何将此 json 解析为行？

Answer 1

在 JSON 中使用 ID 作为键是一个非常糟糕的主意，因为这意味着您的结构一直在变化。理想情况下，键应该保持不变，只是值必须改变。 但是你可以用MapType作弊。

您需要使用此架构在 spark 中正确获取 JSON 数据：

from pyspark.sql import types as T

schm = T.StructType(
    [
        T.StructField("code", T.IntegerType()),
        T.StructField("msg", T.StringType()),
        T.StructField(
            "sensors",
            T.MapType(  # This map type allows any ID as key
                T.StringType(),
                T.StructType(
                    [
                        T.StructField("fieldname", T.StringType()),
                        T.StructField("info", T.StringType()),
                        T.StructField("id", T.IntegerType()),
                        T.StructField(
                            "work",
                            T.MapType(  # Same here
                                T.StringType(),
                                T.StructType(
                                    [
                                        T.StructField("ts", T.IntegerType()),
                                        T.StructField("te", T.IntegerType()),
                                        T.StructField("lat", T.StringType()),
                                        T.StructField("lon", T.StringType()),
                                    ]
                                ),
                            ),
                        ),
                    ]
                ),
            ),
        ),
    ]
)

那么，你的数据框应该有这个架构：

df.printSchema()
root
 |-- code: integer (nullable = true)
 |-- msg: string (nullable = true)
 |-- sensors: map (nullable = true)
 |    |-- key: string
 |    |-- value: struct (valueContainsNull = true)
 |    |    |-- fieldname: string (nullable = true)
 |    |    |-- info: string (nullable = true)
 |    |    |-- id: integer (nullable = true)
 |    |    |-- work: map (nullable = true)
 |    |    |    |-- key: string
 |    |    |    |-- value: struct (valueContainsNull = true)
 |    |    |    |    |-- ts: integer (nullable = true)
 |    |    |    |    |-- te: integer (nullable = true)
 |    |    |    |    |-- lat: string (nullable = true)
 |    |    |    |    |-- lon: string (nullable = true)

然后，只需使用explode 和select，就可以实现所需的输出：

df.select(F.explode("sensors").alias("key", "sensor")).select(
    "sensor", F.explode("sensor.work").alias("key", "work")
).select(
    "sensor.id",
    "sensor.fieldname",
    "sensor.info",
    "work.te",
    "work.ts",
    "work.lat",
    "work.lon",
).show()

+------+-------------+--------------+----------+----------+----------+---------+
|    id|    fieldname|          info|        te|        ts|       lat|      lon|
+------+-------------+--------------+----------+----------+----------+---------+
|258443|DeviceStatus2|Unlock Request|1623021124|1623020526|-25.911997|28.169119|
|258443|DeviceStatus2|Unlock Request|1623021453|1623021393|-25.911997|28.169119|
|258443|DeviceStatus2|Unlock Request|1623052783|1623052154|-25.790348|28.308369|
|258441|         Dis1|          Door|1623035978|1623021447|-25.911997|28.169119|
|258441|         Dis1|          Door|1623058925|1623052775|-25.790348|28.308369|
|258442|  AnalogInput|     Lock Fire|1623021633|1623021453|-25.911997|28.169119|
|258442|  AnalogInput|     Lock Fire|1623052962|1623052783|-25.790348|28.308369|
+------+-------------+--------------+----------+----------+----------+---------+