如何将 Seq[T] 列添加到包含两个数据集元素的数据集？答案

【问题标题】：How to add a Seq[T] column to a Dataset that contains elements of two Datasets?如何将 Seq[T] 列添加到包含两个数据集元素的数据集？
【发布时间】：2019-03-24 01:07:59
【问题描述】：

我有两个数据集 AccountData 和 CustomerData，以及相应的案例类：

case class AccountData(customerId: String, forename: String, surname: String)

customerId|accountId|balance|
+----------+---------+-------+
|   IND0002|  ACC0002|    200|
|   IND0002|  ACC0022|    300|
|   IND0003|  ACC0003|    400|
+----------+---------+-------+


case class CustomerData(customerId: String, accountId: String, balance: Long)
+----------+-----------+--------+
|customerId|   forename| surname|
+----------+-----------+--------+
|   IND0001|Christopher|   Black|
|   IND0002|  Madeleine|    Kerr|
|   IND0003|      Sarah| Skinner|
+----------+-----------+--------+

如何派生以下数据集，其中添加包含每个 customerId 的 Seq[AccountData] 的列 accounts？

+----------+-----------+----------------------------------------------+
|customerId|forename   |surname   |accounts                           |                                               
+----------+-----------+----------+---------------------------------- +
|IND0001   |Christopher|Black     |[]                                                                   
|IND0002   |Madeleine  |Kerr      |[[IND0002,ACC002,200],[IND0002,ACC0022,300]]                        
|IND0003   |Sarah      |Skinner   |[[IND0003,ACC003,400]

我试过了：

val joinCustomerAndAccount =  accountDS.joinWith(customerDS, customerDS("customerId") === accountDS("customerId")).drop(col("_2"))

这给了我以下数据框：

+---------------------+
|_1                   |
+---------------------+
|[IND0002,ACC0002,200]|
|[IND0002,ACC0022,300]|
|[IND0003,ACC0003,400]|
+---------------------+

如果我这样做：

val result = customerDS.withColumn("accounts", joinCustomerAndAccount("_1")(0))

我得到以下异常：

Exception in thread "main" org.apache.spark.sql.AnalysisException: Field name should be String Literal, but it's 0;

【问题讨论】：

标签： scala apache-spark apache-spark-sql apache-spark-dataset

【解决方案1】：

帐户可以按“customerId”分组并与客户一起加入：

// data
val accountDS = Seq(
  AccountData("IND0002", "ACC0002", 200),
  AccountData("IND0002", "ACC0022", 300),
  AccountData("IND0003", "ACC0003", 400)
).toDS()

val customerDS = Seq(
  CustomerData("IND0001", "Christopher", "Black"),
  CustomerData("IND0002", "Madeleine", "Kerr"),
  CustomerData("IND0003", "Sarah", "Skinner")
).toDS()

// action
val accountsGroupedDF = accountDS.toDF
  .groupBy("customerId")
  .agg(collect_set(struct("accountId", "balance")).as("accounts"))

val result = customerDS.toDF.alias("c")
  .join(accountsGroupedDF.alias("a"), $"c.customerId" === $"a.customerId", "left")
    .select("c.*","accounts")

result.show(false)

输出：

+----------+-----------+-------+--------------------------------+
|customerId|forename   |surname|accounts                        |
+----------+-----------+-------+--------------------------------+
|IND0001   |Christopher|Black  |null                            |
|IND0002   |Madeleine  |Kerr   |[[ACC0002, 200], [ACC0022, 300]]|
|IND0003   |Sarah      |Skinner|[[ACC0003, 400]]                |
+----------+-----------+-------+--------------------------------+

【讨论】：