Spark DataFrame按列值聚合去重整理（无UDF）求助

输入DataFrame #

+----------------------------------------------+-------------------------------------+
|                        market_Year           |       Quarter-Range                 |
+----------------------------------------------+-------------------------------------+
|/equity/dividends[DIV|4Y]                     |    null
|/equity/dividends[DIV|1Y]                     |    [0D-6M]                          |
|/equity/dividends[DIV|1Y]                     |    [0D-6M],[6M-18M]                 |
|/equity/dividends[DIV|2Y]                     |    [18M-2Y]                         |
|/equity/dividends[DIV|2Y]                     |    [18M-2Y]                         |
|/equity/dividends[DIV|3Y]                     |    [18M-2Y]                         |
|/equity/dividends[DIV|2Y]                     |    [18M-2Y]                         |
|/equity/dividends[DIV|1Y]                     |    [6M-18M]                         |
+----------------------------------------------+-------------------------------------+

期望输出DataFrame #

+----------------------------------------------+-------------------------------------+
|                        market_Year           |                Quarter-Range        |
+----------------------------------------------+-------------------------------------+
|/equity/dividends[DIV|1Y]                     |  [0D-6M],[6M-18M]                   |
|/equity/dividends[DIV|2Y]                     |  [18M-2Y]                           |
|/equity/dividends[DIV|3Y]                     |  [18M-2Y]                           |
|/equity/dividends[DIV|4Y]                     |                                     |
+----------------------------------------------+-------------------------------------+

尝试的代码及问题 #

尝试了以下代码：

Dataset<Row> DF_to = printdf.withColumn("Quarter-Range", explode(array("Quarter-Range")))
                .groupBy(col("market_Year"))
                .agg(collect_set("Quarter-Range").distinct());

但得到了重复元素的数组输出，格式也不符合预期：

+----------------------------------------------+-------------------------------------+
|                        market_Year           |          Quarter-Range              |
+----------------------------------------------+-------------------------------------+
|/equity/dividends[DIV|1Y]                     |[[0D-6M],[6M-18M], [0D-6M], [6M-18M]]|
|/equity/dividends[DIV|2Y]                     |                           [[18M-2Y]]|
|/equity/dividends[DIV|3Y]                     |                           [[18M-2Y]]|
|/equity/dividends[DIV|4Y]                     |                                   []|
+----------------------------------------------+-------------------------------------+

解决方案 #

核心思路是先拆分多值的Quarter-Range，再去重合并，全程使用Spark内置函数：

1. 替换null为空字符串，避免后续处理报错；
2. 按逗号拆分Quarter-Range为数组，再展开每个元素；
3. 过滤掉空元素，避免收集到无效值；
4. 分组后用collect_set去重收集元素；
5. 用concat_ws将数组转为逗号分隔的字符串，空数组则显示空字符串。

对应的Java代码：

import static org.apache.spark.sql.functions.*;

Dataset<Row> resultDF = printdf
    // 替换null为空字符串，拆分多值为数组
    .withColumn("Quarter-Range", split(coalesce(col("Quarter-Range"), lit("")), ","))
    // 展开数组中的每个元素
    .withColumn("Quarter-Range", explode(col("Quarter-Range")))
    // 过滤掉空元素（处理原null或空字符串拆分后的空值）
    .filter(trim(col("Quarter-Range")).notEqual(""))
    // 按market_Year分组，收集去重后的元素
    .groupBy(col("market_Year"))
    .agg(collect_set("Quarter-Range").alias("Quarter-Range"))
    // 将数组转为逗号分隔的字符串，空数组转空字符串
    .withColumn("Quarter-Range", concat_ws(",", col("Quarter-Range")));

代码说明 #

• coalesce(col("Quarter-Range"), lit(""))：把null替换成空字符串，防止split处理null报错；
• split(..., ",")：将逗号分隔的多值拆分成数组；
• explode：把数组的每个元素拆成单独行；
• filter(trim(...).notEqual(""))：过滤掉拆分后可能出现的空元素；
• collect_set：自动去重收集元素，返回无重复的数组；
• concat_ws(",", ...)：把数组用逗号连接成字符串，数组为空时返回空字符串。

内容的提问来源于stack exchange，提问作者usr90