Spark的dropDuplicates或distinct 对数据去重

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

消除重复的数据可以通过使用 distinct 和 dropDuplicates 两个方法。

distinct数据去重

distinct 是所有的列进行去重的操作,假如你的 DataFrame里面有10列,那么只有这10列完全相同才会去重。
使用distinct:返回当前DataFrame中不重复的Row记录。该方法和接下来的dropDuplicates()方法不传入指定字段时的结果相同。

dropDuplicates()y有四个重载方法

  1. 第一个def dropDuplicates(): Dataset[T] = dropDuplicates(this.columns)
    这个方法,不需要传入任何的参数,默认根据所有列进行去重,然后按数据行的顺序保留每行数据出现的第一条。
 def dropDuplicates(colNames: Array[String]): Dataset[T] = dropDuplicates(colNames.toSeq)
  1. 第二个def dropDuplicates(colNames: Seq[String])
    传入的参数是一个序列。你可以在序列中指定你要根据哪些列的重复元素对数据表进行去重,然后也是返回每一行数据出现的第一条
def dropDuplicates(colNames: Seq[String]): Dataset[T] = withTypedPlan {
    val resolver = sparkSession.sessionState.analyzer.resolver
    val allColumns = queryExecution.analyzed.output
    val groupCols = colNames.toSet.toSeq.flatMap { (colName: String) =>
      // It is possibly there are more than one columns with the same name,
      // so we call filter instead of find.
      val cols = allColumns.filter(col => resolver(col.name, colName))
      if (cols.isEmpty) {
        throw new AnalysisException(
          s"""Cannot resolve column name "$colName" among (${schema.fieldNames.mkString(", ")})""")
      }
      cols
    }
    Deduplicate(groupCols, planWithBarrier)
  }
  1. 第三个def dropDuplicates(colNames: Array[String])
    传入的参数是一个数组,然后方法会把数组转换为序列然后再调用第二个方法。
def dropDuplicates(colNames: Array[String]): Dataset[T] = dropDuplicates(colNames.toSeq)
  1. 第四个def dropDuplicates(col1: String, cols: String*)
    传入的参数为字符串,在方法体内会把你传入的字符串组合成一个序列再调用第二个方法。
  @scala.annotation.varargs
  def dropDuplicates(col1: String, cols: String*): Dataset[T] = {
    val colNames: Seq[String] = col1 +: cols
    dropDuplicates(colNames)
  }

第三和第四个本质上还是调用了第二个方法,所以我们在使用的时候如果需要根据指定的列进行数据去重,可以直接传入一个Seq。

第一个方法默认根据所有列去重,实际上也是调用了第二个方法,然后传入参数this.columns,即所有的列组成的Seq。

所以各位想深究dropDuplicate()去重的核心代码,只需要研究第二个去重方法即可。

特别的

使用dropDuplicate时,可以先排序,去重时,会保留排第一的,如:

val sourceByTime = df.select("url", "name","updatetime").repartition(col("url"))//这一步是为了解决下面注意事项里出现的问题
            .orderBy(desc("updatetime")).dropDuplicates(Seq("url")) 

上面代码,会保留每个url最新的数据。

注意事项

用dropDuplicates可能会出现重复数据,原因:

数据存在多个excuter中,因为spark是分布式计算的,数据在计算的时候会分布在不同的excutor上,使用dropDuplicate去重的时候,可能只是一个excutor内的数据进行了去重,别的excutor上可能还会有重复的数据。

数据是存放在不同分区的,因为spark是分布式计算的,数据在计算的时候会分散在不同的分区中,使用dropDuplicate去重的时候,不同的区分可能还会存在相同的数据。

参考

https://www.cnblogs.com/Jaryer/p/13558701.html
https://blog.csdn.net/qq_39900031/article/details/115797287
https://www.yangch.net/archives/114文章来源地址https://www.toymoban.com/news/detail-599595.html

到了这里,关于Spark的dropDuplicates或distinct 对数据去重的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • spark sql 数据倾斜--join 同时开窗去重的问题优化

    背景: 需求:在一张查询日志表中,有百亿数据,需要join上维表,再根据几个字段进行去重 开窗去重和join 一定要分步进行 ,按照需求先做join再开窗,或者去重完成后在进行join。 dwd_tmp1 中存在百亿用户查询日志数据 数据倾斜 数据量超百亿,资源给到200 * 2c * 20G,执行引擎

    2024年02月11日
    浏览(52)
  • count distinct在spark中的运行机制

    假设源数据分布在两个1核的结点上,数据就8行 spark把count distinct操作转换成count操作。 第一步是对每个要count distinct的列,生成新的行(这里是product和category列),当然原来不需要distinct聚合的列也在。 原来items列不需要distinct,product和category列要distinct,所以数据膨胀了2倍。

    2024年02月04日
    浏览(36)
  • 大数据Spark SparkSession的3种创建方式 Scala语言实现

    SparkSession是Apache Spark 2.0版本引入的一个编程接口,用于与Spark进行交互。它是Spark应用程序的入口点,提供了一种方便的方式来创建DataFrame、DataSet和SQLContext等数据结构,并且可以配置各种Spark应用程序的选项。SparkSession还管理了Spark应用程序的运行环境,包括Spark集群的连接,

    2023年04月20日
    浏览(36)
  • Spark大数据分析与实战笔记(第一章 Scala语言基础-2)

    Spark是专为大规模数据处理而设计的快速通用的计算引擎,它是由Scala语言开发实现的,关于大数据技术,本身就是计算数据,而Scala既有面向对象组织项目工程的能力,又具备计算数据的功能,同时Spark和Scala的紧密集成,本书将采用Scala语言开发Spark程序,所以学好Scala将有助

    2024年02月11日
    浏览(59)
  • Spark大数据分析与实战笔记(第一章 Scala语言基础-1)

    Spark是专为大规模数据处理而设计的快速通用的计算引擎,它是由Scala语言开发实现的,关于大数据技术,本身就是计算数据,而Scala既有面向对象组织项目工程的能力,又具备计算数据的功能,同时Spark和Scala的紧密集成,本书将采用Scala语言开发Spark程序,所以学好Scala将有助

    2024年02月11日
    浏览(62)
  • Spark大数据分析与实战笔记(第一章 Scala语言基础-3)

    对于每一门编程语言来说,数组(Array)都是重要的数据结构之一,主要用来存储数据类型相同的元素。Scala中的数组分为定长数组和变长数组,定义定长数组,需要使用new,而定义变长数组时,则需要导包 import scala.collection.mutable.ArrayBuffer 。 数组(Array)主要用来存储

    2024年02月10日
    浏览(60)
  • Kafka传输数据到Spark Streaming通过编写程序java、scala程序实现操作

    现有一电商网站数据文件,名为buyer_favorite1,记录了用户对商品的收藏数据,数据以“t”键分割,数据内容及数据格式如下: 项目环境说明 开启hadoop集群,zookeeper服务,开启kafka服务。再另开启一个窗口,在/apps/kafka/bin目录下创建一个topic。 1、新创一个文件folder命名为li

    2024年02月13日
    浏览(55)
  • Linux CentOS下大数据环境搭建(zookeeper+hadoop+hbase+spark+scala)

    本篇文章是结合我个人学习经历所写,如果遇到什么问题或者我有什么错误,欢迎讨论。 百度网盘链接:https://pan.baidu.com/s/1DCkQQVYqYHYtPws9hWGpgw?pwd=zh1y 提取码:zh1y 软件在连接中VMwareWorkstation_V16.2.1_XiTongZhiJia的文件夹下。 双击运行安装包,这里下一步即可。 这里勾选我接受许可

    2024年04月15日
    浏览(66)
  • 构建大数据环境:Hadoop、MySQL、Hive、Scala和Spark的安装与配置

    在当今的数据驱动时代,构建一个强大的大数据环境对于企业和组织来说至关重要。本文将介绍如何安装和配置Hadoop、MySQL、Hive、Scala和Spark,以搭建一个完整的大数据环境。 安装Hadoop 首先,从Apache Hadoop的官方网站下载所需的Hadoop发行版。选择适合你系统的二进制发行版,下

    2024年02月11日
    浏览(50)
  • 大数据平台安装实验: ZooKeeper、Kafka、Hadoop、Hbase、Hive、Scala、Spark、Storm

    ​ 在大数据时代,存在很多开源的分布式数据采集、计算、存储技术,本实验将在熟练掌握几种常见Linux命令的基础上搭建几种常用的大数据采集、处理分析技术环境。 相关安装包下载: 链接:https://pan.baidu.com/s/1Wa2U3qstc54IAUCypcApSQ 提取码:lcd8 Hadoop大数据平台所需工具、软件

    2023年04月09日
    浏览(88)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包