scala - 将多个单独的条目合并为 Spark Dataframe 中的单个条目

Question

假设我有一个看起来像这样的分区

part1:

{"customerId":"1","name":"a"}
{"customerId":"2","name":"b"}

假设我想将其架构更改为类似

{"data":"customers":[{"customerId":"1","name":"a"},{"customerId":"2","name":"b"}]}

我尝试做的是

case class Customer(customerId:Option[String],name:Option[String])
case class Customers(customers:Option[Seq[Customer]])
case class Datum(data:Option[Customers])

我尝试将分区读取为 Json 并转换为 Dataframe。

val inputJson = spark.read.format("json").load("part1")
inputJson.as[Datum]

Dataframe 似乎并没有隐式地推断模式。

Answer 1

通过这种结构，我相信您隐藏/包装了数据中真正有用的信息。这里唯一有用的信息是:{"customerId":"1","name":"a"},{"customerId":"2","name":"b"} 客户连同数据只会隐藏您真正需要的数据。为了立即访问数据，您必须先将数据略微更改为:

{"customers":[{"customerId":"1","name":"a"},{"customerId":"2","name":"b"}]}

然后使用下一段代码访问此 JSON:

case class Customer(customerId:String, name:String)
case class Data(customers: Array[Customer])

val df = spark.read.json(path).as[Data]

如果尝试打印此数据框，您会得到:

+----------------+
|       customers|
+----------------+
|[[1, a], [2, b]]|
+----------------+

这当然是将您的数据包装到数组中。现在到了有趣的部分，为了访问它，您必须执行以下操作:

df.foreach{ data => data.customers.foreach(println _) }

这将打印:

Customer(1,a)
Customer(2,b)

这是您需要但根本不容易访问的真实数据。

编辑:

我不使用 2 个类，而是只使用一个类，即 Customer 类。然后利用内置的 Spark 过滤器来选择内部 JSON 对象。最后，您可以展开每个客户数组，并从展开的列生成一个 Customer 类的强类型数据集。

这是最终代码:

case class Customer(customerId:String, name:String)

val path = "C:\\temp\\json_data.json"
val df = spark.read.json(path)

df.select(explode($"data.customers"))
  .map{ r => Customer(r.getStruct(0).getString(0), r.getStruct(0).getString(1))}
  .show(false)

输出:

+----------+----+
|customerId|name|
+----------+----+
|1         |a   |
|2         |b   |
+----------+----+