时间戳

首页 > 解决方案 > 在读取/加载时保留原始 JSON 作为 Spark DataFrame 中的列？

json - 在读取/加载时保留原始 JSON 作为 Spark DataFrame 中的列？

问题描述

在将数据读入 Spark DataFrame 时，我一直在寻找一种将原始 (JSON) 数据添加为列的方法。我有一种方法可以通过加入来做到这一点，但我希望有一种方法可以在使用 Spark 2.2.x+ 的单个操作中做到这一点。

例如数据：

{"team":"Golden Knights","colors":"gold,red,black","origin":"Las Vegas"}
{"team":"Sharks","origin": "San Jose", "eliminated":"true"}
{"team":"Wild","colors":"red,green,gold","origin":"Minnesota"}

执行时：

val logs = sc.textFile("/Users/vgk/data/tiny.json") // example data file
spark.read.json(logs).show

可以预见的是，我们得到：

+--------------+----------+--------------------+--------------+
|        colors|eliminated|              origin|          team|
+--------------+----------+--------------------+--------------+
|gold,red,black|      null|           Las Vegas|Golden Knights|
|          null|      true|            San Jose|        Sharks|
|red,green,gold|      null|           Minnesota|          Wild|
|red,white,blue|     false|District of Columbia|      Capitals|
+--------------+----------+--------------------+--------------+

我希望在初始加载时拥有上述内容，但将原始 JSON 数据作为附加列。例如（截断的原始值）：

+--------------+-------------------------------+--------------+--------------------+
|        colors|eliminated|              origin|          team|               value|
+--------------+----------+--------------------+--------------+--------------------+
|red,white,blue|     false|District of Columbia|      Capitals|{"colors":"red,wh...|
|gold,red,black|      null|           Las Vegas|Golden Knights|{"colors":"gold,r...|
|          null|      true|            San Jose|        Sharks|{"eliminated":"tr...|
|red,green,gold|      null|           Minnesota|          Wild|{"colors":"red,gr...|
+--------------+----------+--------------------+--------------+--------------------+

非理想解决方案涉及连接：

val logs = sc.textFile("/Users/vgk/data/tiny.json")
val df = spark.read.json(logs).withColumn("uniqueID",monotonically_increasing_id)
val rawdf = df.toJSON.withColumn("uniqueID",monotonically_increasing_id)
df.join(rawdf, "uniqueID")

这会产生与上面相同的数据框，但带有并添加了uniqueID列。此外，json 是从 DF 呈现的，不一定是“原始”数据。在实践中它们是相等的，但对于我的用例，实际的原始数据更可取。

是否有人知道将原始 JSON 数据捕获为加载时的附加列的解决方案？

标签： jsonapache-sparkapache-spark-sql

解决方案

如果您有收到的数据的架构，那么您可以使用from_jsonwithschema来获取所有字段并保持raw字段原样

val logs = spark.sparkContext.textFile(path) // example data file

val schema = StructType(
  StructField("team", StringType, true)::
  StructField("colors", StringType, true)::
  StructField("eliminated", StringType, true)::
  StructField("origin", StringType, true)::Nil
)

logs.toDF("values")
    .withColumn("json", from_json($"values", schema))
    .select("values", "json.*")

    .show(false)

输出：

+------------------------------------------------------------------------+--------------+--------------+----------+---------+
|values                                                                  |team          |colors        |eliminated|origin   |
+------------------------------------------------------------------------+--------------+--------------+----------+---------+
|{"team":"Golden Knights","colors":"gold,red,black","origin":"Las Vegas"}|Golden Knights|gold,red,black|null      |Las Vegas|
|{"team":"Sharks","origin": "San Jose", "eliminated":"true"}             |Sharks        |null          |true      |San Jose |
|{"team":"Wild","colors":"red,green,gold","origin":"Minnesota"}          |Wild          |red,green,gold|null      |Minnesota|
+------------------------------------------------------------------------+--------------+--------------+----------+---------+

希望他的帮助！

推荐阅读