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的所有字段传进去。

参考

https://blog.csdn.net/qq_39900031/article/details/115797287