apache-spark - 将 JSON 字符串列拆分为多列

Question

我正在寻找一种通用解决方案，以将所有 json 字段提取为 JSON 字符串列中的列。

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

'path'中文件的文件格式为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 字段。但实际上，我无法遍历所有行来找到所有字段。

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

Answer 1

如果它是一个 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}

代码

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|
+--------------+---+----+----+