Spark原理——逻辑执行图

逻辑执行图

  • 明确逻辑计划的边界

    在 Action 调用之前，会生成一系列的RDD,这些RDD之间的关系，其实就是整个逻辑计划

    val conf= new SparkConf().setMaster("local[6]").setAppName("wordCount_source")
val sc= new SparkContext(conf)val textRDD=sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val splitRDD=textRDD.flatMap(_.split(" "))
val tupleRDD=splitRDD.map((_, 1))
val reduceRDD=tupleRDD.reduceByKey(_ + _)
val strRDD=reduceRDD.map(item => s"${item._1},${item._2}")println(strRDD.toDebugString)
strRDD.collect.foreach(item =>println(item))

    例如上述代码，如果生成逻辑计划的，会生成效如下一些RDD,这些RDD是相互关联的，这些RDD之间，其实本质上生成的就是一个 计算链

    在这里插入图片描述

    接下来, 采用迭代渐进式的方式, 一步一步的查看一下整体上的生成过程

  • RDD 如何生成

    • **textFile算子的背后**

      研究 RDD 的功能或者表现的时候, 其实本质上研究的就是 RDD中的五大属性, 因为 RDD透过五大属性来提供功能和表现, 所以如果要研究 textFile 这个算子, 应该从五大属性着手, 那么第一步就要看看生成的 RDD是什么类型的 RDD

      1. **textFile 生成的是 HadoopRDD**

        在这里插入图片描述

        在这里插入图片描述

        在这里插入图片描述

      2. **HadoopRDD 的 Partitions对应了 HDFS的 Blocks**

        在这里插入图片描述

        其实本质上每个 HadoopRDD的 Partition都是对应了一个 Hadoop的 Block, 通过 InputFormat 来确定 Hadoop中的 Block 的位置和边界, 从而可以供一些算子使用

        在这里插入图片描述

      3. HadoopRDD 的 compute函数就是在读取 HDFS中的 Block

        本质上, compute 还是依然使用 InputFormat来读取 HDFS中对应分区的 Block

      4. textFile这个算子生成的其实是一个 MapPartitionsRDD

        textFile这个算子的作用是读取 HDFS上的文件, 但是 HadoopRDD中存放是一个元组, 其 Key是行号, 其 Value是 Hadoop中定义的 Text对象, 这一点和 MapReduce程序中的行为是一致的

        但是并不适合 Spark 的场景, 所以最终会通过一个 map算子, 将 (LineNum, Text)转为 String形式的一行一行的数据, 所以最终 textFile这个算子生成的 RDD 并不是 HadoopRDD, 而是一个 MapPartitionsRDD

    • map算子的背后

      在这里插入图片描述

      • map算子生成了 MapPartitionsRDD

         val conf= new SparkConf().setMaster("local[6]").setAppName("wordCount_source")
val sc= new SparkContext(conf)val rdd=sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val rdd1=rdd.flatMap(_.split(" "))
val rdd2=rdd1.map((_, 1))

        由源码可知, 当 val rdd2 = rdd1.map() 的时候, 其实生成的新 RDD 是 rdd2, rdd2 的类型是 MapPartitionsRDD, 每个 RDD 中的五大属性都会有一些不同, 由 map 算子生成的 RDD 中的计算函数, 本质上就是遍历对应分区的数据, 将每一个数据转成另外的形式

      • MapPartitionsRDD的计算函数是 collection.map( function )

        真正运行的集群中的处理单元是 Task, 每个 Task 对应一个 RDD 的分区, 所以 collection 对应一个 RDD 分区的所有数据, 而这个计算的含义就是将一个 RDD 的分区上所有数据当作一个集合, 通过这个 Scala 集合的 map 算子, 来执行一个转换操作, 其转换操作的函数就是传入 map 算子的 function

      • 传入 map算子的函数会被清理

        在这里插入图片描述

        这个清理主要是处理闭包中的依赖, 使得这个闭包可以被序列化发往不同的集群节点运行

    • flatMap算子的背后

      在这里插入图片描述

      flatMap和 map算子其实本质上是一样的, 其步骤和生成的 RDD都是一样, 只是对于传入函数的处理不同, map是 collect.map( function )而 flatMap是 collect.flatMap( function )

      从侧面印证了, 其实 Spark中的 flatMap和 Scala基础中的 flatMap其实是一样的

      val conf= new SparkConf().setMaster("local[6]").setAppName("wordCount_source")
val sc= new SparkContext(conf)val textRDD=sc.parallelize(Seq("Hadoop Spark", "Hadoop Flume", "Spark Sqoop"))
val splitRDD=textRDD.flatMap(_.split(" "))
val tupleRDD=splitRDD.map((_, 1))
val reduceRDD=tupleRDD.reduceByKey(_ + _)
val strRDD=reduceRDD.map(item => s"${item._1},${item._2}")// println(strRDD.toDebugString)
strRDD.collect.foreach(item =>println(item))

      textRDD → splitRDD → tupleRDD

      由 textRDD到 splitRDD再到 tupleRDD的过程, 其实就是调用 map和 flatMap算子生成新的 RDD的过程, 所以如下图所示, 就是这个阶段所生成的逻辑计划

  • RDD 之间有哪些依赖关系

    • 前置说明

      • 什么是RDD之间的依赖关系？

        在这里插入图片描述

        • 什么是关系（依赖关系）？

          从算子视角上来看，splitRDD 通过 map 算子得到了 tupleRDD ,所以 splitRDD 和 tupleRDD 之间的关系是 map, 但是仅仅这样说，会不够全面，从细节上来看，RDD只是数据和关于数据的计算，而具体执这种计算得出结果的是一个神秘的其它组件，所以，这两个 RDD 的关系可以表示为 splitRDD 的数据通过 map 操作，被传入 tupleRDD ,这是它们之间更细化的关系

          但是 RDD 这个概念本身并不是数据容器，数据真正应该存放的地方是 RDD 的分区，所以如果把视角放在数据这一层面上的话，直接讲这两个 RDD 之间有关系是不科学的，应该从这两个 RDD 的分区之间的关系来讨论它们之间的关系

        • 那这些分区之间是什么关系？

          如果仅仅说 splitRDD 和 tupleRDD 之间的话，那它们的分区之间就是一对一的关系

          但是 tupleRDD 到 reduceRDD 呢？tupleRDD 通过算子 reduceByKey 生成 reduceRDD ,而这个算子是一个 Shuffle 操作，Shuff1e 操作的两个 RDD 的分区之间并不是一对一，reduceByKey 的一个分区对应 tupleRDD 的多个分区

      • reduceByKey 算子会生成 ShuffledRDD

        reduceByKey 是由算子 combineByKey 来实现的，combineByKey 内部会创建 ShuffledRDD 返回，而整个reduceByKey操作大致如下过程

        在这里插入图片描述

        去掉两个 reducer 端分区，只留下一个的话，如下

        在这里插入图片描述

        所以, 对于 reduceByKey 这个 Shuffle 操作来说, reducer 端的一个分区, 会从多个 mapper 端的分区拿取数据, 是一个多对一的关系

        至此为止, 出现了两种分区见的关系了, 一种是一对一, 一种是多对一

    • 窄依赖

      窄依赖（NarrowDependency）

      假如 rddB = rddA.transform(…), 如果 rddB 中一个分区依赖 rddA 也就是其父 RDD 的少量分区, 这种 RDD 之间的依赖关系称之为窄依赖

      换句话说, 子 RDD 的每个分区依赖父 RDD 的少量个数的分区, 这种依赖关系称之为窄依赖

      @Test
def narrowDependency(): Unit = {// 需求：求得两个 RDD 之间的笛卡尔积// 1. 生成 RDDval conf = new SparkConf().setMaster("local[6]").setAppName("cartesian")val sc = new SparkContext(conf)val rddA = sc.parallelize(Seq(1, 2, 3))val rddB = sc.parallelize(Seq("a", "b"))// 2. 计算val rddC = rdd1.cartesian(rdd2)// 3. 结果获取rddC.collect().foreach(print(_))sc.stop()
}
// 运行结果： (1,a) (1,b) (2,a) (2,b) (3,a) (3,b)
      • 上述代码的 cartesian 是求得两个集合的笛卡尔积
      • 上述代码的运行结果是 rddA 中每个元素和 rddB 中的所有元素结合, 最终的结果数量是两个 RDD 数量之和
      • rddC 有两个父 RDD, 分别为 rddA 和 rddB

      对于 cartesian 来说, 依赖关系如下

      在这里插入图片描述

      上述图形中清晰展示如下现象

      • rddC 中的分区数量是两个父 RDD 的分区数量之乘积
      • rddA 中每个分区对应 rddC 中的两个分区 (因为 rddB 中有两个分区), rddB 中的每个分区对应 rddC 中的三个分区 (因为 rddA 有三个分区)

      它们之间是窄依赖, 事实上在 cartesian 中也是 NarrowDependency 这个所有窄依赖的父类的唯一一次直接使用, 为什么呢?

      因为所有的分区之间是拷贝关系, 并不是 Shuffle 关系

      • rddC 中的每个分区并不是依赖多个父 RDD 中的多个分区
      • rddC 中每个分区的数量来自一个父 RDD 分区中的所有数据, 是一个 FullDependence, 所以数据可以直接从父 RDD 流动到子 RDD
      • 不存在一个父 RDD 中一部分数据分发过去, 另一部分分发给其它的 RDD

    • 宽依赖

      宽依赖（ShuffleDependency

      （并没有所谓的宽依赖, 宽依赖应该称作为 ShuffleDependency）

      在 ShuffleDependency 的类声明上如下写到

      Represents a dependency on the output of a shuffle stage.

      上面非常清楚的说道, 宽依赖就是 Shuffle 中的依赖关系, 换句话说, 只有 Shuffle 产生的地方才是宽依赖

      那么宽窄依赖的判断依据就非常简单明确了, 是否有 Shuffle ?

      举个 reduceByKey 的例子, rddB = rddA.reduceByKey( (curr, agg) ⇒ curr + agg ) 会产生如下的依赖关系

      在这里插入图片描述

      在这里插入图片描述

      • rddB 的每个分区都几乎依赖 rddA 的所有分区
      • 对于 rddA 中的一个分区来说, 其将一部分分发给 rddB 的 p1, 另外一部分分发给 rddB 的 p2, 这不是数据流动, 而是分发

    • 如何分辨宽窄依赖 ?

      其实分辨宽窄依赖的本身就是在分辨父子 RDD 之间是否有 Shuffle, 大致有以下的方法

      • 如果是 Shuffle, 两个 RDD 的分区之间不是单纯的数据流动, 而是分发和复制
      • 一般 Shuffle 的子 RDD 的每个分区会依赖父 RDD 的多个分区

      先看是否一对一>是就是窄依赖，如果不是一对一，是多对一>不能确定，再继续判断

      但是这样判断其实不准确, 如果想分辨某个算子是否是窄依赖, 或者是否是宽依赖, 则还是要取决于具体的算子, 例如想看 cartesian 生成的是宽依赖还是窄依赖, 可以通过如下步骤

      1. 查看 map 算子生成的 RDD

        在这里插入图片描述

        在这里插入图片描述

        在这里插入图片描述

      2. 进去 RDD 查看 getDependence 方法

        在这里插入图片描述

        在这里插入图片描述

        在这里插入图片描述

        在这里插入图片描述
        724024159.png?origin_url=Untitled%2520180.png&pos_id=img-mqipe2i4-1704890196048)

    • 常见的窄依赖类型

      Dependency.scala 源码有。

      在这里插入图片描述

      • 一对一窄依赖

        其实 RDD 中默认的是 OneToOneDependency, 后被不同的 RDD 子类指定为其它的依赖类型, 常见的一对一依赖是 map 算子所产生的依赖, 例如 rddB = rddA.map(…)

        在这里插入图片描述

        • 每个分区之间一一对应, 所以叫做一对一窄依赖

      • Range 窄依赖

        Range 窄依赖其实也是一对一窄依赖, 但是保留了中间的分隔信息, 可以通过某个分区获取其父分区, 目前只有一个算子生成这种窄依赖, 就是 union 算子, 例如 rddC = rddA.union(rddB)

        在这里插入图片描述

        • rddC 其实就是 rddA 拼接 rddB 生成的, 所以 rddC 的 p5 和 p6 就是 rddB 的 p1 和 p2
        • 所以需要有方式获取到 rddC 的 p5 其父分区是谁, 于是就需要记录一下边界, 其它部分和一对一窄依赖一样

      • 多对一窄依赖

        多对一窄依赖其图形和 Shuffle 依赖非常相似, 所以在遇到的时候, 要注意其 RDD 之间是否有 Shuffle 过程, 比较容易让人困惑, 常见的多对一依赖就是重分区算子 coalesce, 例如 rddB = rddA.coalesce(2, shuffle = false), 但同时也要注意, 如果 shuffle = true 那就是完全不同的情况了

        在这里插入图片描述

        • 因为没有 Shuffle, 所以这是一个窄依赖

      • 再谈宽窄依赖的区别

        • 宽窄依赖的区别非常重要, 因为涉及了一件非常重要的事情: 如何计算 RDD ?
        • 宽窄依赖的核心区别是: 窄依赖的 RDD 可以放在一个 Task 中运行

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/623573.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

JVM篇--Java内存区域高频面试题

JVM篇--Java内存区域高频面试题

java内存区域 1 Java 堆空间及 GC？ 首先我们要知道java堆空间的产生过程： 即当通过java命令启动java进程的时候，就会为它分配内存，而分配内存的一部分就会用于创建堆空间，而当程序中创建对象的时候 就会从堆空间来分…
阅读更多...
2024--Django平台开发-Redis集群（十一）

2024--Django平台开发-Redis集群（十一）

内容回顾 主从复制。 哨兵：实例启动了，哨兵节点没启动，Python通过redis-py连接报错。一定要确保实例节点和哨兵节点都启动了。 搭建集群用的是虚拟机的多台centos服务器，你在跟着学习的时候，一定要全部都是虚拟机&am…
阅读更多...
5.矩阵分析

5.矩阵分析

矩阵分析 文章目录 矩阵分析一、方阵范数1.1 矩阵范数1.2 与矩阵乘积相容的矩阵范数【定义】自相容范数 / 方阵范数 1.3 与向量范数相容的矩阵范数【定义】矩阵范数与向量范数相容【定理】任意自相容范数必存在与它相容的向量范数 二、算子范数2.1 方阵的算子范数【定理】由向量…
阅读更多...
PVE虚拟机安装qemu guest agent

PVE虚拟机安装qemu guest agent

pve虚拟机安装guest agent，使web平台可以直接显示虚拟机的ip，方便管理。 一、虚拟机需开启Qemu代理 首先，虚拟机需开启Qemu代理，需要关闭虚拟机再启动虚拟机并安装agent。网上有些文章说要把网卡配置为virtio，经测试是…
阅读更多...
可协作植物管理工具HortusFox

可协作植物管理工具HortusFox

什么是 HortusFox ？ HortusFox 是一个自托管的协作植物管理系统，您可以在自己的环境中使用它来管理所有的植物。您可以添加植物，并附带各种详细信息和照片，并将它们分配到环境中的位置。系统提供了一个仪表板，显示所有…
阅读更多...
小程序系列--6.全局配置

小程序系列--6.全局配置

一. 全局配置文件及常用的配置项 二、window 1. 小程序窗口的组成部分 2. 了解 window 节点常用的配置项 3. 设置导航栏的标题 4. 设置导航栏的背景色 5. 设置导航栏的标题颜色 6. 全局开启下拉刷新功能 7. 设置下拉刷新时窗口的背景色 8. 设置下拉刷新时 loading 的样…
阅读更多...
Git 使用与问题记录 二（公司快速上手版）

Git 使用与问题记录 二（公司快速上手版）

写在前面 记录自己学习的内容，方便后面忘记的时候查看。给像我一样的新手提供一点参考 正文 上一章已经安装好了Git，如何使用呢。我这里会分享两种办法，第一种是在VS2022中克隆代码，修改和提交；第二种是用命令提交。…
阅读更多...
【计算机网络】TCP原理 | 可靠性机制分析（四）

【计算机网络】TCP原理 | 可靠性机制分析（四）

个人主页：兜里有颗棉花糖 欢迎 点赞👍 收藏✨ 留言✉ 加关注💓本文由 兜里有颗棉花糖 原创 收录于专栏【网络编程】 本专栏旨在分享学习计算机网络的一点学习心得，欢迎大家在评论区交流讨论💌 这里写目录标题 &#x1…
阅读更多...
C函数详解 | 函数的作用、定义与声明、函数的调用、函数与指针

C函数详解 | 函数的作用、定义与声明、函数的调用、函数与指针

欢迎关注博主 Mindtechnist 或加入【智能科技社区】一起学习和分享Linux、C、C、Python、Matlab，机器人运动控制、多机器人协作，智能优化算法，滤波估计、多传感器信息融合，机器学习，人工智能等相关领域的知识和技术。关…
阅读更多...
RISC-V是如何与X86、ARM三分天下

RISC-V是如何与X86、ARM三分天下

目录 1.行业CPU指令集格局 2.汽车中的RISC-V进展 2.1 国际进展 2.2 国内进展 3.小结 2023年3月2日，在平头哥牵头举办的玄铁RISC-V生态大会上，工程院院士倪光南表示，基于RISC-V模块化、可扩展、容易定制、不受垄断制约等优势，…
阅读更多...
电商API接口|Javascript抓取京东、淘宝商品数据

电商API接口|Javascript抓取京东、淘宝商品数据

“ 不知怎么建站？就找怎么建站！ ” 背景： EDI许可证网站和ICP许可证网站需要有丰富的商品数据来应付EDI、ICP许可证下证审核。下面介绍的这种方法是我之前主要的抓取数据的方法，大概用了一年多。这几天又对这个方法进行了一些优…
阅读更多...
11.云原生分布式数据库之TIDB

11.云原生分布式数据库之TIDB

云原生专栏大纲 文章目录 为什么使用TIDB后端视角运维视角基础架构视角 TiDB Operator 简介软件版本要求部署tidbTIDB工具helm常用命令TIDB学习推荐资料 为什么使用TIDB 从后端视角、运维视角和基础架构视角来看，使用 TiDB 作为数据库系统可以获得分布式架构、高可…
阅读更多...
学习Java API（二）：基础知识点一文通✅

学习Java API（二）：基础知识点一文通✅

推荐阅读 智能化校园：深入探讨云端管理系统设计与实现（一） 智能化校园：深入探讨云端管理系统设计与实现（二） 文章目录 推荐阅读正则表达式正则表达式语法捕获组和分组matches 和 lookingAt 方法split方法r…
阅读更多...
LeeCode前端算法基础100题（20）找出字符串中第一个匹配项的下标

LeeCode前端算法基础100题（20）找出字符串中第一个匹配项的下标

一、问题详情： 给你两个字符串 haystack 和 needle ，请你在 haystack 字符串中找出 needle 字符串的第一个匹配项的下标（下标从 0 开始）。如果 needle 不是 haystack 的一部分，则返回 -1 。 示例 1： 输入：haystack = "sadbutsad", needle = "s…
阅读更多...
安全生产管理平台——革新传统安全生产管理方式，重塑企业安全文化

安全生产管理平台——革新传统安全生产管理方式，重塑企业安全文化

安全生产管理在现代企业中占据着至关重要的地位。传统的安全生产管理方式虽然在一定程度上能够保障企业的生产安全，但随着企业规模的不断扩大和生产环境的日益复杂，其局限性也愈发凸显。而安全生产管理平台的出现，正是为了解决这一问题。 平台…
阅读更多...
从六西格玛到DFSS：提升企业创新能力的关键方法——张驰咨询

从六西格玛到DFSS：提升企业创新能力的关键方法——张驰咨询

DFSS（Design for Six Sigma，即六西格玛设计）是一种结合了六西格玛思想和创新设计方法的质量管理方法。它的目标是在产品或服务的设计阶段，通过优化设计过程和减少变异性，以确保产品或服务达到高质量水平，满…
阅读更多...
Linux多网卡绑定实现负载均衡详解

Linux多网卡绑定实现负载均衡详解

将多块网卡绑定同一IP地址对外提供服务，可以实现高可用或者负载均衡。直接给两块网卡设置同一IP地址是不可以的。通过 bonding，虚拟一块网卡对外提供连接，物理网卡的被修改为相同的MAC地址。 目录 1、bond的作用 2、Bonding聚合链路工作模…
阅读更多...
反射助你无痛使用Semantic Kernel接入离线大模型

反射助你无痛使用Semantic Kernel接入离线大模型

本文主要介绍如何使用 llama 的 server 部署离线大模型，并通过反射技术修改 Semantic Kernel 的 OpenAIClient 类，从而实现指定端点的功能。最后也推荐了一些学习 Semantic Kernel 的资料，希望能对你有所帮助。 封面图片： Dalle3 …
阅读更多...
软信天成：数据安全管理解决方案分享

软信天成：数据安全管理解决方案分享

近年来，随着数据环境日趋复杂多变和潜在的数据隐私泄露风险潜伏，如何确保企业数据安全已成为众多企业亟待面对与妥善处理的重要问题。 为了应对这一严峻的现实挑战，软信天成凭借专业的知识体系和丰富的实战经验积累，总结出了一套…
阅读更多...
Java多线程并发篇----第十四篇

Java多线程并发篇----第十四篇

系列文章目录 文章目录 系列文章目录前言一、ReadWriteLock 读写锁二、共享锁和独占锁三、重量级锁（Mutex Lock）四、轻量级锁前言 前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。点击跳转到网站，这篇文章男女通用，看懂了就去分享给…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023