合并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]

某种程度上，数据框似乎并没有隐含地推断该架构。

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)

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

编辑：

而不是使用两个类，而是只使用一个客户类。然后利用建筑物火花过滤器来选择内部JSON对象。最后，您可以爆炸每个客户数组，并从爆炸的列生成一个强大类型的类客户的数据集。

这是最终代码：

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   |
+----------+----+

Answer 2

我最终操纵了数据帧本身

val inputJson = spark.read.format("json").load("part1")
val formatted = inputJson.withColumn("dummy",lit(1)).groupBy("dummy")
.agg(collect_list(struct(dataFrame.col("*"))).alias("customers"))
val finalFormatted=formatted.withColumn("data",struct(col("customers")))
.select("data")

现在我做

finalFormatted.printSchema

我得到了我需要的模式

  |-- data: struct (nullable = false)
  |    |-- customers: array (nullable = true)
  |    |    |-- element: struct (containsNull = true)
  |    |    |    |-- customerId: string (nullable = true)
  |    |    |    |-- name: string (nullable = true)