spark merge datasets based on the same input of one column and concat the others

user_12772599 0
Viewed 67

Currently I have several Dataset[UserRecord], and it looks like this

case class UserRecord(
  Id: String,
  ts: Timestamp,
  detail: String
)

Let's call the several datasets datasets. Previously I tried this

datasets.reduce(_ union _)
  .groupBy("Id")
  .agg(collect_list("ts", "detail"))
  .as[(String, Seq[DetailRecord]]

but this code gives me an OOM error. I think the root cause is collect_list.

Now I'm thinking if I can do the groupBy and agg for each of the dataset first and then join them together to solve the OOM issue. Any other good advice is welcome too :)

I have an IndexedSeq of datasets look like this

|name|   lists      |
| x  |[[1,2], [3,4]]|

|name|   lists      |
| y  |[[5,6], [7,8]]|

|name|   lists         |
| x  |[[9,10], [11,12]]|

How can I combine them to get a Dataset that looks like

|name|          lists               |
| x  |[[1,2], [3,4],[9,10], [11,12]]|
| y  |[[5,6], [7,8]]                |

I tried ds.reduce(_ union _) but it didn't seem to work

apache-spark scala
1 Answers

You can aggregate after union:

val ds2 = ds.reduce(_ unionAll _).groupBy("name").agg(flatten(collect_list("lists")).as("lists"))
user_141657300
Related