如何合并火花数据集中的行以组合字符串列答案

【问题标题】：How to merging rows in a spark data set to combine a string column如何合并火花数据集中的行以组合字符串列
【发布时间】：2021-06-26 03:27:52
【问题描述】：

我需要将数据集中的两行或多行合并为一行。必须根据 id 列进行分组。要合并的列是一个字符串。我需要在合并列中获取一个逗号分隔的字符串。我如何实现这一点是Java。输入行

col1,col2  
1,abc  
2,pqr  
1,abc1  
3,xyz
2,pqr1

预期输出：

col1, col2  
1, "abc,abc1"  
2, "pqr,pqr1"  
3, xyz

【问题讨论】：

请分享数据、代码和预期输出。
已编辑以添加预期的示例
顺便说一句，您的示例无法阅读

标签： apache-spark

【解决方案1】：

聚合两个单独的列：

your_data_frame
    .withColumn("aggregated_column", concat_ws(",", col("col1"), col("col2"))

以防万一，这里是除了通常的东西之外要导入的东西

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

编辑

如果你想聚合任意数量的你知道名字的列，你可以这样做：

String[] column_names = {"c1", "c2", "c3"};
Column[] columns = Arrays.asList(column_names)
            .stream().map(x -> col(x))
            .collect(Collectors.toList())
            .toArray(new Column[0]);
data_frame
    .withColumn("agg", concat_ws(",", columns));

编辑 #2：分组和连接

如果您想按“ID”列分组并聚合另一列，您可以这样做：

dataframe
    .groupBy("ID")
    .agg(concat_ws(",", collect_list(col("col1")) ))

【讨论】：

myDS.groupBy("ID").agg(functions.concat_ws(",",myDS.col("THEME")));这就是我所做的，但我看到一个例外，它说表达式 THEME 既不在 groupBy 中，也不在聚合中。我想从 2 个或多个具有相同 ID 值的不同行中聚合同一列。e
concat_ws 不是聚合函数。这就是您收到此错误的原因。据我所知，唯一的解决方案是使用 functions.collect_list，它将您的值聚合到一个列表中，然后应用 UDF（用户定义的函数）。
好吧，事实证明没有 udf 也是可能的。 concat_ws 可以使用字符串数组列。我为我的答案添加了一个解决方案。

【解决方案2】：

使用 groupBy 和 concat_ws

import org.apache.spark.sql.functions._
df.groupBy("col1").agg(concat_ws(",", collect_list("col2")))

【讨论】：