如何从 Spark SQL 中的逗号分隔字符串中删除重复项？

Question

这是我的数据集的示例。我想编写 Spark SQL 以将具有重复值的项目列表更改为唯一值，例如“apple,banana,banana”将变为“apple,banana” 我在这里写了代码：

SELECT  ItemId, Date, concat_ws(',',collect_set(CAST(SPLIT(PointNameArray, ','),ARRAY<STRING>))) AS VarN
FROM    dataset 

但我不断收到错误提示

无法解析 'concat_ws(',', collect_set(split(dataset.PointNameArray, ',')))' 由于数据类型 不匹配：参数 2 需要（数组或字符串）类型，但是， 'collect_set(split(dataset.PointNameArray, ','))' 是 数组

itemlist 的结果是

apple, banana
phone, computer
table, chair

所以表格应该是这样的

Answer 1

你得到了错误：

无法解析 'concat_ws(',', collect_set(split(dataset.PointNameArray, ',')))' 由于数据类型 不匹配：参数 2 需要（数组或字符串）类型，但是， 'collect_set(split(dataset.PointNameArray, ','))' 是数组

因为split 已经返回一个array<string> 并且collect_set 将整个数组视为集合中的一个条目，因此array<array<string>>。

需要注意的是，collect_set 是一个聚合函数，可以按组应用，也可以作为每行的窗口函数应用。然而，将它作为窗口函数应用在每一行上会冒着将整个拆分数组视为一个元素的风险，而这不是目标。

此外，当您split, 的一些样本数据（'apple, banana, banana）时，有些可能在值前有空格（ ）， banana 与banana 不同。为了确保我们拥有独特的价值观，我们可以trim这些。

由于您使用的是 spark sql，因此下面的示例将使用 spark-sql 而不是可以遵循相同方法的 python/scala api

之前的示例数据框

+------+----------+---------------------+
|ItemId|Date      |itemList             |
+------+----------+---------------------+
|item1 |2019-01-01|apple, banana, banana|
+------+----------+---------------------+

建议的代码

SELECT  
    ItemId, 
    Date, 
    concat_ws(',',collect_set(trim(isplit))) as VarN
FROM
    df
LATERAL VIEW explode(SPLIT(itemList,',')) as isplit
GROUP BY ItemId, Date

之后的输出

+------+----------+------------+
|ItemId|Date      |VarN        |
+------+----------+------------+
|item1 |2019-01-01|apple,banana|
+------+----------+------------+

上面的示例代码中的数据是：

1. 由, 使用SPLIT(itemList,',') 拆分
2. 在侧视图输出中使用explode(SPLIT(itemList,',')) 分解成多行

+------+----------+-------+
|ItemId|Date      |isplit |
+------+----------+-------+
|item1 |2019-01-01|apple  |
|item1 |2019-01-01| banana|
|item1 |2019-01-01| banana|
+------+----------+-------+

1. 然后使用trim(isplit) 修剪数据以删除空格
2. 然后我们使用collect_set 聚合数据，这将提供我们的唯一值，并在其他列ItemId, Date 上进行分组。注意。如果您有很多列，您可以在唯一列上进行分组，例如 ItemId，并在其他列上简单地使用 MAX，例如

SELECT  
    ItemId, 
    MAX(Date) as Date, 
    concat_ws(',',collect_set(trim(isplit))) as VarN
FROM
    df
LATERAL VIEW explode(SPLIT(itemList,',')) as isplit
GROUP BY ItemId

5. 最后我们使用concat_ws连接这些值

Answer 2

我认为你想使用array_distinct 函数而不是collect_set：

spark.sql("""
SELECT  ItemId, 
        Date, 
        PointNameArray,
        concat_ws(',',array_distinct(split(PointNameArray, ','))) AS VarN
FROM    dataset
""").show(truncate=False)

#+------+----------+-------------------------+--------------+
#|ItemId|Date      |PointNameArray           |VarN          |
#+------+----------+-------------------------+--------------+
#|item1 |2019-01-01|apple,banana,apple,banana|apple,banana  |
#|item2 |2019-01-01|phone,computer,computer  |phone,computer|
#|item3 |2019-01-01|table,chair,table,chair  |table,chair   |
#+------+----------+-------------------------+--------------+