将SQL CSV火花转换为具有不同数据类型的JSON

Question

目前，我有这样的csv数据：

id,key,value
id_1,int_key,1
id_1,string_key,asd
id_1,double_key,null
id_2,double_key,2.0

我想将按其ID和相应的正确数据类型分组的这些属性转换为json。

我期望具有这样的json结构：

[{
  id: "id_1"
  attributes: {
    int_key: 1,
    string_key: "asd"
    double_key: null
  }
},
  id: "id_2"
  attributes: {
    double_key: 2.0
  }]

我当前的解决方案是在Spark中使用to_json的collect_list，如下所示：

SELECT to_json(id, map_from_arrays(collect_list(key), collect_list(value)) as attributes GROUP BY id)

但是，这行得通，我找不到一种方法转换为正确的数据类型。

[{
  id: "id_1"
  attributes: {
    int_key: "1",
    string_key: "asd"
    double_key: "null"
  }
},
  id: "id_2"
  attributes: {
    double_key: "2.0"
  }]

我还需要添加对空值的支持。但是我已经找到了解决方案。我在to_json中使用ignoreNulls选项。因此，如果我尝试枚举每个属性并将其强制转换为它们对应的类型，则将包括所有已定义的属性。我只想包含csv文件中定义的用户的属性。

顺便说一句，我正在使用Spark 2.4。

Answer 1

Python：这是我从scala版本转换的PySpark版本。结果是相同的。

from pyspark.sql.functions import col, max, struct

df = spark.read.option("header","true").csv("test.csv")
keys = [row.key for row in df.select(col("key")).distinct().collect()]

df2 = df.groupBy("id").pivot("key").agg(max("value"))
df2.show()
df2.printSchema()

for key in keys:
    df2 = df2.withColumn(key, col(key).cast(key.split('_')[0]))
df2.show()
df2.printSchema()

df3 = df2.select("id", struct("int_key", "double_key", "string_key").alias("attributes"))
jsonArray = df3.toJSON().collect()

for json in jsonArray: print(json)

---

Scala：我尝试通过首先使用pivot来拆分每种类型的值。

val keys = df.select('key).distinct.rdd.map(r => r(0).toString).collect
val df2 = df.groupBy('id).pivot('key, keys).agg(max('value))
df2.show
df2.printSchema

然后，DataFrame如下所示：

+----+-------+----------+----------+
|  id|int_key|double_key|string_key|
+----+-------+----------+----------+
|id_2|   null|       2.0|      null|
|id_1|      1|      null|       asd|
+----+-------+----------+----------+

root
 |-- id: string (nullable = true)
 |-- int_key: string (nullable = true)
 |-- double_key: string (nullable = true)
 |-- string_key: string (nullable = true)

其中每列的类型仍然是字符串。要进行投射，我使用了foldLeft，

val df3 = keys.foldLeft(df2) { (df, key) => df.withColumn(key, col(key).cast(key.split("_").head)) }
df3.show
df3.printSchema

并且结果现在具有正确的类型。

+----+-------+----------+----------+
|  id|int_key|double_key|string_key|
+----+-------+----------+----------+
|id_2|   null|       2.0|      null|
|id_1|      1|      null|       asd|
+----+-------+----------+----------+

root
 |-- id: string (nullable = true)
 |-- int_key: integer (nullable = true)
 |-- double_key: double (nullable = true)
 |-- string_key: string (nullable = true)

然后，您可以构建您的json，例如

val df4 = df3.select('id, struct('int_key, 'double_key, 'string_key) as "attributes")
val jsonArray = df4.toJSON.collect
jsonArray.foreach(println)

最后一行用于检查结果的位置

{"id":"id_2","attributes":{"double_key":2.0}}
{"id":"id_1","attributes":{"int_key":1,"string_key":"asd"}}