Extracting array elements and mapping them to case class

Question

The following is the output I am getting after performing a groupByKey, mapGroups and then a joinWith operation on the caseclass dataset:

+------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------+
|_1                            |_2                                                                                                                                                        |
+------------------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------+
|[IND0001,Christopher,Black]   |null                                                                                                                                                            |
|[IND0002,Madeleine,Kerr]      |[IND0002,WrappedArray([IND0002,ACC0155,323], [IND0002,ACC0262,60])]                                                                                             |
|[IND0003,Sarah,Skinner]       |[IND0003,WrappedArray([IND0003,ACC0235,631], [IND0003,ACC0486,400], [IND0003,ACC0540,53])]                                                                      |
|[IND0004,Rachel,Parsons]      |[IND0004,WrappedArray([IND0004,ACC0116,965])]                                                                                                                   |
|[IND0005,Oliver,Johnston]     |[IND0005,WrappedArray([IND0005,ACC0146,378], [IND0005,ACC0201,34], [IND0005,ACC0450,329])]                                                                      |
|[IND0006,Carl,Metcalfe]       |[IND0006,WrappedArray([IND0006,ACC0052,57], [IND0006,ACC0597,547])]                                                                                             |

The code is as follows:

val test = accountDS.groupByKey(_.customerId).mapGroups{ case (id, xs) => (id, xs.toSeq)}
  test.show(false)

val newTest = customerDS.joinWith(test, customerDS("customerId") === test("_1"), "leftouter")
newTest.show(500,false)

Now I want to take the arrays and output them in a format as follows:

 +----------+-----------+----------+---------------------------------------------------------------------+--------------+------------+-----------------+
 * |customerId|forename   |surname   |accounts                                                             |numberAccounts|totalBalance|averageBalance   |
 * +----------+-----------+----------+---------------------------------------------------------------------+--------------+------------+-----------------+
 * |IND0001   |Christopher|Black     |[]                                                                   |0             |0           |0.0              |
 * |IND0002   |Madeleine  |Kerr      |[[IND0002,ACC0155,323], [IND0002,ACC0262,60]]                        |2             |383         |191.5            |
 * |IND0003   |Sarah      |Skinner   |[[IND0003,ACC0235,631], [IND0003,ACC0486,400], [IND0003,ACC0540,53]] |3             |1084        |361.3333333333333|

Note: I cannot use spark.sql.functions._ at all --> training academy rules :(

How do I get the above output which should be mapped to a case class as follows:

case class CustomerAccountOutput(
                                    customerId: String,
                                    forename: String,
                                    surname: String,
                                    //Accounts for this customer
                                    accounts: Seq[AccountData],
                                    //Statistics of the accounts
                                    numberAccounts: Int,
                                    totalBalance: Long,
                                    averageBalance: Double
                                  )

I really need help with this. Stuck with this for weeks without a working solution.

user_12400214 · Answer

Let's say you have the following DataFrame:

val sourceDf = Seq((1, Array("aa", "CA")), (2, Array("bb", "OH"))).toDF("id", "data_arr")
sourceDf.show()

// output:
+---+--------+
| id|data_arr|
+---+--------+
|  1|[aa, CA]|
|  2|[bb, OH]|
+---+--------+

and you want to convert it to the following schema:

val destSchema = StructType(Array(
  StructField("id", IntegerType, true),
  StructField("name", StringType, true),
  StructField("state", StringType, true)
))

You can do:

val destDf: DataFrame = sourceDf
.map { sourceRow =>
  Row(sourceRow(0), sourceRow.getAs[mutable.WrappedArray[String]](1)(0), sourceRow.getAs[mutable.WrappedArray[String]](1)(1))
}(RowEncoder(destSchema))

destDf.show()

// output:
+---+----+-----+
| id|name|state|
+---+----+-----+
|  1|  aa|   CA| 
|  2|  bb|   OH|
+---+----+-----+

Extracting array elements and mapping them to case class

1 Answers