Spark Sql之dropDuplicates去重-Toy模板网

这篇具有很好参考价值的文章主要介绍了Spark Sql之dropDuplicates去重。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

算子介绍

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

distinct源码：

def distinct(): Dataset[T] = dropDuplicates()

def dropDuplicates(): Dataset[T] = dropDuplicates(this.columns)

可以看到底层调用的是dropDuplicates(this.columns)
而columns: Array[String] = schema.fields.map(_.name)
其实就是获取df的所有字段传进去。