spark group by 去重

### 使用 `groupBy()` 实现去重操作 在 Apache Spark 中，`groupBy()` 可以用来对 DataFrame 的行进行分组[^1]。然而，如果目标是对数据执行去重操作，则可以结合其他方法来完成此任务。 #### 方法一：使用 `dropDuplicates()` 虽然这不是严格意义上的 `groupBy()` 去重方式，但在大多数情况下，直接调用 `dropDuplicates()` 是更高效的选择。它可以基于指定列或整个 DataFrame 来删除重复项： ```python from pyspark.sql import SparkSession # 创建 SparkSession spark = SparkSession.builder.appName("example").getOrCreate() data = [("Alice", 20), ("Bob", 30), ("Alice", 20)] columns = ["name", "age"] df = spark.createDataFrame(data, columns) # 删除重复记录 distinct_df = df.dropDuplicates(["name"]) distinct_df.show() ``` 上述代码会保留每种名称的第一条记录，并移除其余相同的记录。 --- #### 方法二：通过 `groupBy().agg()` 进行去重 当需要利用 `groupBy()` 和聚合函数实现去重时，可以选择某些特定字段作为键值来进行分组，并从中提取唯一的数据组合。例如： ```python from pyspark.sql.functions import first result_df = df.groupBy("name").agg(first("age").alias("age")) result_df.show() ``` 这里通过对 `"name"` 列进行分组，并选取第一个对应的年龄值，从而实现了按名字维度的去重效果。 需要注意的是，在实际应用过程中可能会涉及 Shuffle 操作，这通常是由于分组或者聚合引起的[^2]。因此对于大规模数据集来说，性能优化可能成为一个考虑因素。 --- #### 方法三：借助 `combineByKey()` 自定义逻辑 另一种更为灵活的方式是采用低层次 API —— `combineByKey()` ，该方法允许开发者自定义创建初始状态以及更新规则的方法[^4]^5]。下面展示了一个简单的例子说明如何运用它来做基本形式上的“去重”。 假设有一个由 `(key,value)` 对组成的 RDD 数据源: ```scala val rdd = sc.parallelize(Array(("a",1),("b",2),("a",3))) // 定义三个参数分别为初始化器、累加器和合并器 rdd.combineByKey( v => Set(v), (acc:Set[Int],v) => acc + v, (set1:Set[Int], set2:Set[Int]) => set1 ++ set2 ).collect.foreach(println) ``` 在这个案例里，我们把所有的 value 放入到一个不可变集合(Set)当中，这样自然就达到了去除相同元素的目的。 --- ### 性能考量与注意事项 - **Shuffle 开销**: 当涉及到诸如 `groupBy`, `reduceByKey` 等算子时，不可避免会产生 shuffle 动作，这对资源消耗较大尤其是处理海量数据场景下应格外注意其影响程度。 - **内存占用**: 如果 keys 数量巨大可能导致过多的小文件生成进而增加 GC 负担等问题发生；此时可适当调整分区数目加以缓解压力。