将 JSON 字符串列拆分为多列

我正在寻找一个通用的解决方案来从 JSON 字符串列中提取所有 json 字段作为列。

df =  spark.read.load(path)
df.show()

“路径”中文件的文件格式是 parquet

样本数据

|id | json_data
| 1 | {"name":"abc", "depts":["dep01", "dep02"]}
| 2 | {"name":"xyz", "depts":["dep03"],"sal":100}
| 3 | {"name":"pqr", "depts":["dep02"], "address":{"city":"SF","state":"CA"}}

预期产出

|id | name    | depts              | sal | address_city | address_state
| 1 | "abc"   | ["dep01", "dep02"] | null| null         | null
| 2 | "xyz"   | ["dep03"]          | 100 | null         | null
| 3 | "pqr"   | ["dep02"]          | null| "SF"         | "CA"

我知道我可以通过创建定义了架构的 StructType 并使用“from_json”方法来提取列。

但这种方法需要手动模式定义。

val myStruct = StructType(
  Seq(
    StructField("name", StringType),
    StructField("depts", ArrayType(StringType)),
    StructField("sal", IntegerType)
  ))

var newDf = df.withColumn("depts", from_json(col("depts"), myStruct))

有没有更好的方法来展平 JSON 列而无需手动定义架构？ 在提供的示例中，我可以看到可用的 JSON 字段。 但实际上，我无法遍历所有行来查找所有字段。

因此，我正在寻找一种解决方案，将所有字段拆分为列，而不指定列的名称或类型。

如果它是一个CSV文件，只有一列作为JSON数据。您可以使用以下解决方案。

val csvDF = spark.read.option("delimiter", "|").option("inferSchema", true).option("header", true).csv("test.csv")
val rdd = csvDF.select(" json_data").rdd.map(_.getString(0))
val ds = rdd.toDS
val jsonDF = spark.read.json(ds)
val jsonDFWithID = jsonDF.withColumn("id", monotonically_increasing_id())
val csvDFWithID = csvDF.select($"id ").withColumn("id", monotonically_increasing_id())
val joinDF = jsonDFWithID.join(csvDFWithID, "id").drop("id")

这就是最终数据框的样子。

scala> joinDF.printSchema()
root
 |-- address: struct (nullable = true)
 |    |-- city: string (nullable = true)
 |    |-- state: string (nullable = true)
 |-- depts: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- name: string (nullable = true)
 |-- sal: long (nullable = true)
 |-- id : double (nullable = true)

如果它是一个，以下解决方案将起作用JSON文件。 为我。inferSchema工作得很好。

json文件

~/Downloads ▶ cat test.json
{"id": 1, "name":"abc", "depts":["dep01", "dep02"]},
{"id": 2, "name":"xyz", "depts" :["dep03"],"sal":100}

code

scala> scc.read.format("json").option("inerSchema", true).load("Downloads/test.json").show()
+--------------+---+----+----+
|         depts| id|name| sal|
+--------------+---+----+----+
|[dep01, dep02]|  1| abc|null|
|       [dep03]|  2| xyz| 100|
+--------------+---+----+----+