java如何获取应用groupby后更改的列值？

Question

我正在使用java编写一个spark代码，在连接条件之后，我们得到了多个记录，因为在不同的源中有重复的IDs，（ID是重复的，但某些属性发生了更改），因此使用相同的id，我们有多个记录。我需要的是将重复的记录合并到每个ID的唯一行上

输入数据集

+---+---+---+----+---+---+
|id |b  |c  |d   |f  |g  |
+---+---+---+----+---+---+
|1  |e  |dd |ddd |34 |r5t|
|1  |e  |dd2|ddd |34 |r5t|
|1  |e  |dd3|ddd |34 |rt |
|2  |e  |dd |ddd1|34 |5rt|
|4  |e  |dd |ddd1|34 |rt |
|1  |e  |dd4|ddd |34 |rt |
|4  |e  |dd4|ddd |34 |rt |
|4  |e  |dd4|ddd |3  |rt |
|2  |e  |dd |ddd |3  |r5t|
|2  |e  |dd |ddd |334|rt |
+---+---+---+----+---+---+

预期产出

+---+--------------+--------------+--------------+-------------------+--------------+
|id |f             |            b |             g|d                  |d             |
+---+--------------+--------------+--------------+-------------------+--------------+
|1  |[34]          |[e]           |[r5t, rt]     |[dd4, dd3, dd2, dd]|[ddd]         |
+---+--------------+--------------+--------------+-------------------+--------------+

我试图明确地给出collect_list，如下所示

df.groupBy("id").agg(


                 functions.collect_set("f"),
                 functions.collect_set("b")

                ).show(1,false);

但在我的例子中，我们有300列，在数据集中，太多的列可能不一样，有时会改变

Answer 1

在Sparkorg.apache.spark.sql包中，有一个agg(exprs: Map[String, String])方法，它接受一个Map<String,String>，其中key是列的名称，值是sql。功能

Dataset<Row> df = spark.read().format("csv").option("header", "true")
                .load("...");

Map<String,String> collect_MAP =  Arrays.stream(df.columns())
            .filter(f -> !f.equals("id"))
            .collect(Collectors.toMap(f -> f,f -> "collect_set"));


     df.groupBy("id").agg(collect_MAP).show(false);

结果

+ -+       +       +       +         -+       +
|id |collect_set(f)|collect_set(b)|collect_set(g)|collect_set(c)     |collect_set(d)|
+ -+       +       +       +         -+       +
|1  |[34]          |[e]           |[r5t, rt]     |[dd4, dd3, dd2, dd]|[ddd]         |
|4  |[3, 34]       |[e]           |[rt]          |[dd4, dd]          |[ddd1, ddd]   |
|2  |[334, 3, 34]  |[e]           |[r5t, rt, 5rt]|[dd]               |[ddd1, ddd]   |
+ -+       +       +       +         -+       +