Spark scala aggregate to an array and concat it

user_17744470 0
Viewed 220

I have a Dataset with a number of columns that looks like this:(Columns -name, timestamp, platform, clickcount, id)

Joy  2021-10-10T10:27:16  apple      5   1
May  2020-12-12T22:28:08  android    6   2
June 2021-09-15T20:20:06  Microsoft  9   3
Joy  2021-09-09T09:30:09  android    10  1
May  2021-08-08T05:05:05  apple      8   2

I want to group by id and after it should look like

Joy  2021-10-10T10:27:16,2021-09-09T09:30:09   apple,android         5,10   1
May  2020-12-12T22:28:08,2021-08-08T05:05:05   android,apple         6,8    2
June 2021-09-15T20:20:06                       Microsoft             9      3

After calling for another Api which converts the id to pseudo id I want to map that id and make it to look like

Joy  2021-10-10T10:27:16,2021-09-09T09:30:09   apple,android         5,10   1   A12
May  2020-12-12T22:28:08,2021-08-08T05:05:05   android,apple         6,8    2   B23
June 2021-09-15T20:20:06                       Microsoft             9      3   C34

I have tried using groupBy and forEach but I am stuck and unable to proceed further

apache-spark aggregate-functions scala
1 Answers

In order to apply the aggregation you want, you should use collect_set as the aggregation function and concat_ws in order to join with comma the created arrays:

import org.apache.spark.sql.functions.{collect_set, concat_ws}
import spark.implicits._

val df: DataFrame = Seq(
  ("joy", "2021-10-10T10:27:16", "apple", 5, 1),
  ("may", "2020-12-12T22:28:08", "android", 6, 2),
  ("june", "2021-09-15T20:20:06", "microsoft", 9, 3),
  ("joy", "2021-09-09T09:30:09", "android", 10, 1),
  ("may", "2021-08-08T05:05:05", "apple", 8, 2)
).toDF("name", "timestamp", "platform", "clickcount", "id")

df
  .groupBy("id")
  .agg(
    concat_ws(",", collect_set("timestamp")).as("timestamp"),
    concat_ws(",", collect_set("name")).as("name"),
    concat_ws(",", collect_set("platform")).as("platform"),
    concat_ws(",", collect_set("clickcount")).as("clickcount")
  ).show()

The output should be:

+---+--------------------+----+-------------+----------+
| id|           timestamp|name|     platform|clickcount|
+---+--------------------+----+-------------+----------+
|  1|2021-10-10T10:27:...| joy|apple,android|      5,10|
|  3| 2021-09-15T20:20:06|june|    microsoft|         9|
|  2|2021-08-08T05:05:...| may|apple,android|       6,8|
+---+--------------------+----+-------------+----------+

In order to add a pseudo id column, you should join the created df with another dataframe that contains the conversion values or write an UDF that will receive an id value and will convert it into pseudo id.

user_124002140
Related