Как получить значения столбцов, которые изменяются после применения groupby? - PullRequest
Новая
       16

Как получить значения столбцов, которые изменяются после применения groupby?

2 голосов
/ 01 октября 2019

Я работаю над кодом зажигания, используя java, где после условия объединения мы получили несколько записей из-за дублирования IDs в разных источниках (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 столбцов, в наборе данных, которые тоже столбцы могут не совпадать, изменяя некоторыераз.

1 Ответ

2 голосов
/ 01 октября 2019

В пакете Spark org.apache.spark.sql есть метод agg(exprs: Map[String, String]), который принимает Map<String,String>, где key - имя столбца, а значение - sql.functions. 

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]   |
+---+--------------+--------------+--------------+-------------------+--------------+
...