java - 删除 Spark 数组列中的重复项

Question

我有一个给定的数据集:

+-------------------+--------------------+
|               date|            products|
+-------------------+--------------------+
|2017-08-31 22:00:00|[361, 361, 361, 3...|
|2017-09-22 22:00:00|[361, 362, 362, 3...|
|2017-09-21 22:00:00|[361, 361, 361, 3...|
|2017-09-28 22:00:00|[360, 361, 361, 3...|

其中 products 列是一个包含可能重复项目的字符串数组。

我想删除这个重复项(在一行内)

我所做的基本上就是像那样编写一个 UDF 函数

 val removeDuplicates: WrappedArray[String] => WrappedArray[String] = _.distinct
 val udfremoveDuplicates = udf(removeDuplicates)

这个解决方案给了我一个正确的结果:

+-------------------+--------------------+--------------------+
|               date|            products|       rm_duplicates|
+-------------------+--------------------+--------------------+
|2017-08-31 22:00:00|[361, 361, 361, 3...|[361, 362, 363, 3...|
|2017-09-22 22:00:00|[361, 362, 362, 3...|[361, 362, 363, 3...|

我的问题是:

1. Spark 是否提供了获得此结果的更好/更有效的方法？

2. 我正在考虑使用 map - 但如何获取所需的列作为列表，以便能够像我的 removeDuplicates lambda 中那样使用“distinct”方法？

编辑:我用 java 标记标记了这个主题，因为对我来说，无论使用哪种语言(scala 或 java)我都会得到答复:)Edit2:错别字

Answer 1

问题中提出的方法——使用 UDF——是最好的方法，因为 spark-sql 没有用于唯一化数组的内置原语。

如果您要处理大量数据和/或数组值具有独特的属性，那么值得考虑 UDF 的实现。

WrappedArray.distinct 在幕后构建一个 mutable.HashSet，然后遍历它以构建不同元素的数组。从性能的角度来看，这有两个可能的问题:

1. Scala 的可变集合并不是非常高效，这就是为什么在 Spark 的内部你会发现很多 Java 集合和 while 循环。如果您需要极致的性能，您可以使用更快的数据结构实现您自己的通用 distinct。

2. distinct 的通用实现不会利用数据的任何属性。例如，如果数组平均较小，那么直接构建数组并对重复项进行线性搜索的简单实现可能比构建复杂数据结构的代码执行得更好，尽管理论上 O(n ^2) 复杂性。再比如，如果取值只能是小范围内的数字，或者是小集合中的字符串，则可以通过位集实现统一化。

同样，只有当您的数据量大得离谱时，才应考虑这些策略。您的简单实现几乎适用于所有情况。