Как добавить столбец Seq [T] в набор данных, который содержит элементы двух наборов данных? - PullRequest
Новая
       33

Как добавить столбец Seq [T] в набор данных, который содержит элементы двух наборов данных?

0 голосов
/ 23 марта 2019

У меня есть два набора данных 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|
+----------+-----------+--------+

Как получить следующий набор данных, который добавляет столбец account , содержащий Seq [ AccountData ] каждого customerId ? 

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

1 Ответ

1 голос
/ 23 марта 2019

Счета могут быть сгруппированы по «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]]                |
+----------+-----------+-------+--------------------------------+
...