1

同事们好。我有一个大数据集（大约 237 000 000 行）。有很多列。例如，我需要删除所有具有列名的重复项userId, VTS。

userId Vts moreColumn1 moreColumn2
10     150     2           3              -delete
11     160     1           6
10     150     0           1              -delete

我对 SQL 不好。尝试了与 Internet 不同的变体，但它不起作用。

更新：

泰求答案！我忘了说我用的是java。有我为java优化的代码：

viewingDataset.groupBy("userId", "VTS")
                .count()
                .where("count = 1")
                .drop("count")
                .join(viewingDataset, JavaConversions.asScalaBuffer(asList("userId", "VTS")))
4

2 回答 2

3

您可以使用计数进行聚合，过滤结果并重新加入

df.groupBy("userId", "Vts").count
  .where($"count" === 1)
  .drop("count")
  .join(df, Seq("userId", "Vts"))

使用窗口函数可以获得相同的结果，但如果数据倾斜并且平均成本更高，则它的鲁棒性会降低。

于 2018-07-04T15:07:11.897 回答
1

您可以使用 Window 函数实现您想要的功能：

import org.apache.spark.sql.expressions.Window._

ds.withColumn("aux", count("*")
.over(Window.partitionBy($"userId", $"VTS")))
.where($"aux"===1)
.drop($"aux")

partitionBy 将根据您作为参数发送的列（在您的示例中为 userId 和 VTS）计算分区有多少元素。然后使用该where子句，我们将只保留计数为 1 的分区中的行，即唯一行。

partitionBY 子句的结果

ds.withColumn("aux", count("*").over(Window.partitionBy($"userId", $"VTS"))).show

+-------+----+------------+------------+---+
| userId| VTS| moreColumn1| moreColumn2|aux|
+-------+----+------------+------------+---+
|     10| 150|           2|           3|  2|
|     10| 150|           0|           1|  2|
|     11| 160|           1|           6|  1|
+-------+----+------------+------------+---+

最后结果

+-------+----+------------+------------+
| userId| VTS| moreColumn1| moreColumn2|
+-------+----+------------+------------+
|     11| 160|           1|           6|
+-------+----+------------+------------+
于 2018-07-04T15:16:15.197 回答