Apache Spark Read JSON с дополнительными столбцами

Question

Я читаю таблицу Hive, в которой есть два столбца: id и jsonString.Я могу легко преобразовать jsonString в структуру данных Spark, вызвав функцию spark.read.json, но я также должен добавить идентификатор столбца.

val jsonStr1 = """{"fruits":[{"fruit":"banana"},{"fruid":"apple"},{"fruit":"pera"}],"bar":{"foo":"[\"daniel\",\"pedro\",\"thing\"]"},"daniel":"daniel data random","cars":["montana","bagulho"]}"""
val jsonStr2 = """{"fruits":[{"dt":"banana"},{"fruid":"apple"},{"fruit":"pera"}],"bar":{"foo":"[\"daniel\",\"pedro\",\"thing\"]"},"daniel":"daniel data random","cars":["montana","bagulho"]}"""
val jsonStr3 = """{"fruits":[{"a":"banana"},{"fruid":"apple"},{"fruit":"pera"}],"bar":{"foo":"[\"daniel\",\"pedro\",\"thing\"]"},"daniel":"daniel data random","cars":["montana","bagulho"]}"""


case class Foo(id: Integer, json: String)

val ds = Seq(new Foo(1,jsonStr1), new Foo(2,jsonStr2), new Foo(3,jsonStr3)).toDS
val jsonDF = spark.read.json(ds.select($"json").rdd.map(r => r.getAs[String](0)).toDS)

jsonDF.show()

jsonDF.show
+--------------------+------------------+------------------+--------------------+
|                 bar|              cars|            daniel|              fruits|
+--------------------+------------------+------------------+--------------------+
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[,,, banana], [,...|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[, banana,,], [,...|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,,,], [,,...|
+--------------------+------------------+------------------+--------------------+

Я хотел бы добавить идентификатор столбца из кустатаблица, как это:

+--------------------+------------------+------------------+--------------------+---------------
|                 bar|              cars|            daniel|              fruits|  id
+--------------------+------------------+------------------+--------------------+--------------
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[,,, banana], [,...|1
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[, banana,,], [,...|2
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,,,], [,,...|3
+--------------------+------------------+------------------+--------------------+

Я не буду использовать регулярные выражения

Я создал udf, который принимает эти два поля в качестве аргумента и использует надлежащую библиотеку JSON includeжелаемое поле (id) и возвращение новой строки JSON, работает как шарм, но я надеюсь, что Spark API предложит лучший способ сделать это.Я использую Apache Spark 2.3.0.

Answer 1

Один из способов - применить from_json к строкам JSON с соответствующей схемой, как показано ниже:

import org.apache.spark.sql.types._
import org.apache.spark.sql.functions._
import spark.implicits._

case class Foo(id: Int, json: String)

val df = Seq(Foo(1, jsonStr1), Foo(2, jsonStr2), Foo(3, jsonStr3)).toDF

val schema = StructType(Seq(
  StructField("bar", StructType(Seq(
    StructField("foo", StringType, true)
    )), true),
  StructField("cars", ArrayType(StringType, true), true),
  StructField("daniel", StringType, true),
  StructField("fruits", ArrayType(StructType(Seq(
    StructField("a", StringType, true),
    StructField("dt", StringType, true),
    StructField("fruid", StringType, true),
    StructField("fruit", StringType, true)
  )), true), true)
))

df.
  withColumn("json_col", from_json($"json", schema)).
  select($"id", $"json_col.*").
  show
// +---+--------------------+------------------+------------------+--------------------+
// | id|                 bar|              cars|            daniel|              fruits|
// +---+--------------------+------------------+------------------+--------------------+
// |  1|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[null,null,null,...|
// |  2|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[null,banana,nul...|
// |  3|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,null,nul...|
// +---+--------------------+------------------+------------------+--------------------+

Answer 2

Я уже знал о функции from_json ранее, но в моем случае было бы "невозможно" вручную вывести схему вручную для каждого JSON. Я думал, что у Spark будет «идиоматический» интерфейс. Это окончательное решение:

ds.select($"id",from_json($"json",jsonDF.schema).alias("_json_path")).select($"_json_path.*",$"id").show

ds.select($"id",from_json($"json",jsonDF.schema).alias("_json_path")).select($"_json_path.*",$"id").show
+--------------------+------------------+------------------+--------------------+---+
|                 bar|              cars|            daniel|              fruits| id|
+--------------------+------------------+------------------+--------------------+---+
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[,,, banana], [,...|  1|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[, banana,,], [,...|  2|
|[["daniel","pedro...|[montana, bagulho]|daniel data random|[[banana,,,], [,,...|  3|
+--------------------+------------------+------------------+--------------------+---+