算子介绍
dropDuplicates去重原则：按数据行的顺序保留每行数据出现的第一条
dropDuplicates 在Spark源码里面提供了以下几个方法重载：
def dropDuplicates(): Dataset[T] = dropDuplicates(this.columns)
这个方法，不需要传入任何的参数，默认根据所有列进行去重，然后按数据行的顺序保留每行数据出现的第一条。
def dropDuplicates(colNames: Seq[String])
传入的参数是一个序列。你可以在序列中指定你要根据哪些列的重复元素对数据表进行去重，然后也是返回每一行数据出现的第一条
def dropDuplicates(colNames: Array[String])
传入的参数是一个数组，然后方法会把数组转换为序列然后再调用第二个方法
def dropDuplicates(col1: String, cols: String*)
传入的参数为字符串，在方法体内会把你传入的字符串组合成一个序列再调用第二个方法。
示例
新建一个dataFrame
val conf = new SparkConf().setAppName("TTyb").setMaster("local")
val sc = new SparkContext(conf)
val spark = new SQLContext(sc)
val dataFrame = spark.createDataFrame(Seq(
(1, 1, "2", "5"),
(2, 2, "3", "6"),
(2, 2, "35", "68"),
(2, 2, "34", "67"),
(2, 2, "38", "68"),
(3, 2, "36", "69"),
(1, 3, "4", null)
)).toDF("id", "label", "col1", "col2")
按照id和lable来去重。此时利用distinct是得不到想要的效果的
利用 dropDuplicates 可以实现
dataFrame.dropDuplicates(Seq("id","label")).show(false)
+---+-----+----+----+
|id |label|col1|col2|
+---+-----+----+----+
|2 |2 |3 |6 |
|1 |1 |2 |5 |
|1 |3 |4 |null|
|3 |2 |36 |69 |
+---+-----+----+----+
问题
问题： 分区多时对导致去重出现问题
需求：求最早时间段的那条数据
原数据：
+---------+----------+
|columnVal|dateStr |
+---------+----------+
|0 |2019-01-01|
|1 |2019-01-01|
|2 |2019-01-01|
|3 |2019-01-01|
|4 |2019-01-01|
|0 |2019-02-01|
|1 |2019-02-01|
|2 |2019-02-01|
|3 |2019-02-01|
|4 |2019-02-01|
|0 |2019-03-01|
|1 |2019-03-01|
|2 |2019-03-01|
|3 |2019-03-01|
|4 |2019-03-01|
+---------+----------+
期望结果：
+---------+----------+
|columnVal|dateStr |
+---------+----------+
|0 |2019-01-01|
|1 |2019-01-01|
|3 |2019-01-01|
|2 |2019-01-01|
|4 |2019-01-01|
+---------+----------+
错误复现：
# 可以看到下面效果跟预期 完全不符合
master("local[4]") //此时分区为4
df.dropDuplicates(Seq("columnVal","dateStr")).show(false)
+---------+----------+
|columnVal|dateStr |
+---------+----------+
|3 |2019-02-01|
|1 |2019-02-01|
|4 |2019-03-01|
|0 |2019-01-01|
|4 |2019-01-01|
|0 |2019-02-01|
|4 |2019-02-01|
|3 |2019-01-01|
|1 |2019-01-01|
|2 |2019-01-01|
|3 |2019-03-01|
|2 |2019-02-01|
|0 |2019-03-01|
|1 |2019-03-01|
|2 |2019-03-01|
+---------+----------+
解决
方式一：
df.orderBy(col("dateStr")) //全局有序 只有一个reduce处理所有数据
.dropDuplicates("columnVal")
.show(false)
方式二：(基于方式一的优化)
df.sortWithinPartitions(col("dateStr")) //每个分区有序
.orderBy(col("dateStr")) //全局有序 只有一个reduce处理所有数据 此时压力会很小
.dropDuplicates("columnVal") //去重
.show(false)
dropDuplicates和distinct
def distinct(): Dataset[T] = dropDuplicates()
def dropDuplicates(): Dataset[T] = dropDuplicates(this.columns)
可以看到底层调用的是dropDuplicates(this.columns)
而columns: Array[String] = schema.fields.map(_.name)
其实就是获取df的所有字段传进去。
参考
https://blog.csdn.net/qq_39900031/article/details/115797287
