在 apache spark 中提取和分解嵌入的 json 字段答案

【问题标题】：Extract and explode embedded json fields in apache spark在 apache spark 中提取和分解嵌入的 json 字段
【发布时间】：2021-06-22 07:07:32
【问题描述】：

我对 spark 完全陌生，但不介意答案是用 python 还是 Scala。出于隐私原因，我无法显示实际数据，但基本上我正在读取具有如下结构的 json 文件：

    {
      "EnqueuedTimeUtc": 'some date time',
      "Properties": {},
      "SystemProperties": {
          "connectionDeviceId": "an id",
          "some other fields that we don't care about": "data"
      },
      "Body": {
        "device_id": "an id",
        "tabs": [
            {
              "selected": false,
               "title": "some title",
               "url": "https:...."
            },
            {"same again, for multiple tabs"}
        ]
     }
   }

大部分数据是不感兴趣的。我想要的是一个由时间、device_id 和 url 组成的 Dataframe。同一设备和时间可以有多个 url，因此我希望将它们分解为每个 url 的一行。

|时间戳 | device_id |网址 |

我的直接问题是，当我读到这篇文章时，虽然它可以计算出 SystemProperties 的结构，但 Body 只是一个字符串，可能是因为变异。也许我需要指定架构，这有帮助吗？

root
 |-- Body: string (nullable = true)
 |-- EnqueuedTimeUtc: string (nullable = true)
 |-- SystemProperties: struct (nullable = true)
 |    |-- connectionAuthMethod: string (nullable = true)
 |    |-- connectionDeviceGenerationId: string (nullable = true)
 |    |-- connectionDeviceId: string (nullable = true)
 |    |-- contentEncoding: string (nullable = true)
 |    |-- contentType: string (nullable = true)
 |    |-- enqueuedTime: string (nullable = true)

有没有一种有效的方法（有很多这样的记录）来提取 url 并与时间和 device_id 相关联？提前致谢。

【问题讨论】：

标签： json apache-spark apache-spark-sql

【解决方案1】：

这是一个提取示例。基本上，您可以使用from_json 将Body 转换为更有条理的内容，并使用explode(transform()) 获取URL 并扩展到不同的行。

# Sample dataframe

df.show(truncate=False)
+----------------------------------------------------------------------------------------------------------------------------------------------+---------------+----------------+
|Body                                                                                                                                          |EnqueuedTimeUtc|SystemProperties|
+----------------------------------------------------------------------------------------------------------------------------------------------+---------------+----------------+
|{"device_id":"an id","tabs":[{"selected":false,"title":"some title","url":"https:1"},{"selected":false,"title":"some title","url":"https:2"}]}|some date time |[an id]         |
+----------------------------------------------------------------------------------------------------------------------------------------------+---------------+----------------+

df.printSchema()
root
 |-- Body: string (nullable = true)
 |-- EnqueuedTimeUtc: string (nullable = true)
 |-- SystemProperties: struct (nullable = true)
 |    |-- connectionDeviceId: string (nullable = true)

# Extract desired properties

df2 = df.selectExpr(
    "EnqueuedTimeUtc as timestamp", 
    "from_json(Body, 'device_id string, tabs array<map<string,string>>') as Body"
).selectExpr(
    "timestamp", 
    "Body.device_id", 
    "explode(transform(Body.tabs, x -> x.url)) as url"
)

df2.show()
+--------------+---------+-------+
|     timestamp|device_id|    url|
+--------------+---------+-------+
|some date time|    an id|https:1|
|some date time|    an id|https:2|
+--------------+---------+-------+

【讨论】：