如何使用 spark/scala 将 json 字符串格式化为 MongoDB 文档样式？

Question

我有一个包含两列 id、json_string 的表，需要将 json_string 转换为 MongoDB 文档格式。我正在将数据从 Spark/Scala 发送到 MongoDB。

我尝试使用 withColumn，但仍然没有得到所需的格式。这是我到目前为止所拥有的，因此非常感谢任何帮助。

原始json字符串样本（df）

val df=spark.sql("select id, json_string from mytable")

{"id":"0001","json_string":"{\"header\": {\"column1\":\"value1\",\"column2\":\"value2\"},\"tail\": [{\"column3\":\"value3\",\"column4\":\"value4\",\"column5\":\"value5\"}]}"}

使用 withColumn (df2) 我得到这个：

val df2=df.withColumn("json_string",from_json(col("json_string"),MapType(StringType,StringType)))

{"id":"0001","json_string":{"header":"{\"column1\":\"value1\",\"column2\":\"value2\"}","tail":"[{\"column3\":\"value3\",\"column4\":\"value4\",\"column5\":\"value5\"}]"}}

所需格式：

{"id":{"$id":"0001"},"header":{"column1":"value1","column2":"value2"},"tail":[{"column3":"value3","column4":"value4","column5":"value5"}]}

Desired format picture sample

Answer 1

您可以动态获取它并将其与from_json 一起使用，而不是手动定义架构

val json_schema = spark.read.json(df.select("json_string").as[String]).schema
val df2 = df.withColumn("json_string", from_json(col("json_string"), json_schema))
  .select("id", "json_string.*")

结果：

+----+----------------+--------------------------+
|id  |header          |tail                      |
+----+----------------+--------------------------+
|0001|{value1, value2}|[{value3, value4, value5}]|
+----+----------------+--------------------------+

架构：

root
 |-- id: string (nullable = true)
 |-- header: struct (nullable = true)
 |    |-- column1: string (nullable = true)
 |    |-- column2: string (nullable = true)
 |-- tail: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- column3: string (nullable = true)
 |    |    |-- column4: string (nullable = true)
 |    |    |-- column5: string (nullable = true)