代码之家 › 专栏 › 技术社区 › Brad Hein

如何使用pyspark explode()爆炸结构

explode pyspark json

Brad Hein · 技术社区 · 5 年前

如何将以下JSON转换为其后的关系行?我坚持的部分是pyspark explode() 由于类型不匹配,函数抛出异常。我还没有找到一种方法将数据强制转换为合适的格式,以便我可以在 source 钥匙内 sample_json 对象。

JSON输入

sample_json = """
{
"dc_id": "dc-101",
"source": {
    "sensor-igauge": {
      "id": 10,
      "ip": "68.28.91.22",
      "description": "Sensor attached to the container ceilings",
      "temp":35,
      "c02_level": 1475,
      "geo": {"lat":38.00, "long":97.00}                        
    },
    "sensor-ipad": {
      "id": 13,
      "ip": "67.185.72.1",
      "description": "Sensor ipad attached to carbon cylinders",
      "temp": 34,
      "c02_level": 1370,
      "geo": {"lat":47.41, "long":-122.00}
    },
    "sensor-inest": {
      "id": 8,
      "ip": "208.109.163.218",
      "description": "Sensor attached to the factory ceilings",
      "temp": 40,
      "c02_level": 1346,
      "geo": {"lat":33.61, "long":-111.89}
    },
    "sensor-istick": {
      "id": 5,
      "ip": "204.116.105.67",
      "description": "Sensor embedded in exhaust pipes in the ceilings",
      "temp": 40,
      "c02_level": 1574,
      "geo": {"lat":35.93, "long":-85.46}
    }
  }
}"""

期望输出

dc_id    source_name    id    description
-------------------------------------------------------------------------------
dc-101   sensor-gauge   10    Sensor attached to the container ceilings
dc-101   sensor-ipad    13    Sensor ipad attached to carbon cylinders
dc-101   sensor-inest    8    Sensor attached to the factory ceilings
dc-101   sensor-istick   5    Sensor embedded in exhaust pipes in the ceilings

PYSPARK代码

from pyspark.sql.functions import *
df_sample_data = spark.read.json(sc.parallelize([sample_json]))
df_expanded = df_sample_data.withColumn("one_source",explode_outer(col("source")))
display(df_expanded)

错误

AnalysisException:无法解析“爆炸”( 来源 )'由于数据类型不匹配:函数爆炸的输入应该是数组或映射类型,而不是结构。。。。

我把这个放在一起了 Databricks notebook 以进一步证明挑战并清楚地显示错误。我将能够使用此笔记本来测试本文提供的任何建议。

0 回复 | 直到 5 年前

blackbishop 5 年前

你不能用 explode 对于结构体,但可以获取结构体中的列名 source (与 df.select("source.*").columns )使用列表理解,您可以从每个嵌套结构中创建所需字段的数组,然后分解以获得所需的结果:

from pyspark.sql import functions as F

df1 = df.select(
    "dc_id",
    F.explode(
        F.array(*[
            F.struct(
                F.lit(s).alias("source_name"),
                F.col(f"source.{s}.id").alias("id"),
                F.col(f"source.{s}.description").alias("description")
            )
            for s in df.select("source.*").columns
        ])
    ).alias("sources")

).select("dc_id", "sources.*") 

df1.show(truncate=False)

#+------+-------------+---+------------------------------------------------+
#|dc_id |source_name  |id |description                                     |
#+------+-------------+---+------------------------------------------------+
#|dc-101|sensor-igauge|10 |Sensor attached to the container ceilings       |
#|dc-101|sensor-inest |8  |Sensor attached to the factory ceilings         |
#|dc-101|sensor-ipad  |13 |Sensor ipad attached to carbon cylinders        |
#|dc-101|sensor-istick|5  |Sensor embedded in exhaust pipes in the ceilings|
#+------+-------------+---+------------------------------------------------+