皮皮网

1.SPARK-38864 - Spark支持unpivot源码分析
2.SparkSQL源码分析-05-SparkSQL的查看从零join处理
3.源码解析Spark中的Parquet高性能向量化读
4.spark原理系列 broadcast广播原理优缺点示例源码权威讲解
5.Spark源码解析2-YarnCluster模式启动
6.Spark-Submit 源码剖析

SPARK-38864 - Spark支持unpivot源码分析

       unpivot是数据库系统中用于列转行的内置函数，如SQL SERVER,源源码 Oracle等。以数据集tb1为例，开启每个数字代表某个人在某个学科的解析成绩。若要将此表扩展为三元组，查看从零可使用union实现。源源码黄柱子买进附图源码但随列数增加，开启SQL语句变长。解析许多SQL引擎提供内置函数unpivot简化此过程。查看从零unpivot使用时需指定保留列、源源码进行转行的开启列、新列名及值列名。解析

       SPARK从SPARK-版本开始支持DataSet的查看从零unpivot函数，逐步扩展至pyspark与SQL。源源码在Dataset API中，开启ids为要保留的Column数组，Column类提供了从String构造Column的隐式转换，方便使用。利用此API，可通过unpivot函数将数据集转换为所需的三元组。values表示转行列，variableColumnName为新列名，valueColumnName为值列名。

       Analyser阶段解析unpivot算子，将逻辑执行计划转化为物理执行计划。当用户开启hive catalog，SPARK SQL根据表名和metastore URL查找表元数据，转化为Hive相关逻辑执行计划。物理执行计划如BroadcastHashJoinExec，表示具体的执行策略。规则ResolveUnpivot将包含unpivot的算子转换为Expand算子，在物理执行计划阶段执行。此转换由开发者自定义规则完成，通过遍历逻辑执行计划树，根据节点类型及状态进行不同处理。

       unpivot函数实现过程中，博星卓越源码首先将原始数据集投影为包含ids、variableColumnName、valueColumnName的列，实现语义转换。随后，通过map函数处理values列，构建新的行数据，最终返回Expand算子。在物理执行计划阶段，Expand算子将数据转换为所需形式，实现unpivot功能。

       综上所述，SPARK内置函数unpivot的实现通过解析列参数，组装Expand算子完成，为用户提供简便的列转行功能。通过理解此过程，可深入掌握SPARK SQL的开发原理与内在机制。

SparkSQL源码分析--SparkSQL的join处理

       SparkSQL的join处理策略多样，针对不同场景各有优劣。首先，map join适用于小表广播至worker节点，提升性能，但大表可能导致OOM。shuffle hash join则对大表进行分区和排序，效率高但内存密集。默认策略通过sort merge join，对大表进行分区排序，避免内存问题，但需预先排序。

       当常规策略不可用时，会考虑等值或不等值join的广播nested loop join，适用于特定条件的right或left outer join。笛卡尔积join在无指定key时使用，仅限inner join。

       SparkPlan中的eclipse 自动关联源码Join子节点与策略紧密相关，如在等值连接时，根据hint选择Broadcast hash join、Shuffle sort merge join或shuffle hash join。没有hint时，依据表大小、join类型和排序情况自动选择。

       非等值连接时，hint会引导使用broadcast nested loop join或Cartesian product join，无hint时则依据表大小和连接类型来决定。

       在特殊情况下，如NotInSubquery，仍可能选择Broadcast hash join。总的来说，SparkSQL的join策略灵活多变，旨在根据具体场景提供最优的执行效率和资源利用率。

源码解析Spark中的Parquet高性能向量化读

       在Spark中，Parquet的高性能向量化读取是自2.0版本开始引入的特性。它与传统的逐行读取和解码不同，采用列式批处理方式，显著提升了列解码的速度，据Databricks测试，速度比非向量化版本快了9倍。本文将深入解析Spark的源码，揭示其如何支持向量化Parquet文件读取。

       Spark的向量化读取主要依赖于ColumnBatch和ColumnVector数据结构。ColumnBatch是每次读取返回的批量数据容器，其中包含一个ColumnVectors数组，每个ColumnVector负责存储一批数据中某一列的所有值。这种设计使得数据可以按列进行高效访问，同时也提供按行的视图，通过InternalRow对象逐行处理。

       在读取过程中，Spark通过VectorizedParquetRecordReader、VectorizedColumnReader和VectorizedValuesReader三个组件协同工作。VectorizedParquetRecordReader负责启动批量读取，asp 多级分类 源码它根据指定的批次大小和内存模式创建实例。VectorizedColumnReader和VectorizedValuesReader则负责实际的列值读取，根据列的类型和编码进行相应的解码处理。

       值得注意的是，Spark在数据加载时会重复使用ColumnBatch和ColumnVector实例，以减少内存占用，优化计算效率。ColumnVector支持堆内存和堆外内存，以适应不同的存储需求。通过这些优化，向量化读取在处理大型数据集时表现出色，尤其是在性能上。

       然而，尽管Spark的向量化读取已经非常高效，Iceberg中的Parquet向量化读取可能更快，这可能涉及到Iceberg对Parquet文件的特定优化，或者其在数据处理流程中的其他改进，但具体原因需要进一步深入分析才能揭示。

spark原理系列 broadcast广播原理优缺点示例源码权威讲解

       Spark广播（broadcast）的原理是通过将一个只读变量从驱动程序发送到集群上的所有工作节点，以便在运行任务时能够高效地访问这个变量。广播变量只会被发送一次，并且在工作节点上缓存，以供后续任务重用。

       这种方式可以避免在任务执行期间多次传输相同的数据，从而提高性能和效率。

       在Spark中，广播变量的实现主要依赖于DriverEndpoint和ExecutorEndpoint之间的通信机制。

       具体来说，当驱动程序将广播变量发送给工作节点时，它会使用BlockManager将序列化的块存储在内存中，并将块的元数据注册到BlockManagerMaster。

       然后，当工作节点执行任务时，它会向BlockManagerMaster请求获取广播变量的块，并从本地BlockManager中获取这些块的炫舞经验源码数据。这样，每个工作节点都可以在本地快速访问广播变量的数据。

       总结起来，Spark广播的实现涉及驱动程序对广播变量进行序列化和发送，以及工作节点接收、反序列化和缓存广播变量的块。这种机制有效地将只读数据分发到集群上的所有工作节点，提高了任务执行的性能和效率。

       广播变量在以下场景中非常有用：

       总之，广播变量适用于需要在多个任务之间共享只读数据，并且能够提供更高效的数据访问和减少网络传输开销的情况。通过使用广播变量，可以提高Spark应用程序的性能和效率。

       虽然广播在分布式计算中有很多优点，但它也存在一些缺点：

       因此，在使用广播变量时需要考虑其局限性和适用场景。如果数据集较大，实时性要求高，或者需要频繁修改数据，可能需要考虑其他替代方案来避免广播的缺点。

       示例源码broadcast方法

       功能：将只读变量广播到集群,返回一个Broadcast对象以在分布式函数中进行读取变量将仅发送一次到每个执行器，同时调用了内部的方法broadcastInternal

       基础类Broadcast抽象类

       Broadcast 是 Spark 中的一个广播变量类。广播变量允许程序员在每台机器上缓存一个只读的变量，而不是将它与任务一起传输。通过使用广播变量，可以以高效的方式为每个节点提供大型输入数据集的副本。

       Broadcast 类的构造函数接收一个唯一标识符 id，用于标识广播变量。

       Broadcast 类是一个抽象类，有以下几个主要方法：

       Broadcast 类还定义了一些受保护的方法，用于实际获取广播变量的值、取消持久化广播变量的值以及销毁广播变量的状态。

       Broadcast 类还具有 _isValid 和 _destroySite 两个私有变量，分别表示广播变量是否有效（即尚未销毁）以及销毁广播变量的位置信息。

       总体来说，Broadcast 类提供了管理广播变量的功能，并确保广播变量的正确使用和销毁。

       实现类TorrentBroadcast

       TorrentBroadcast 是使用类似 BitTorrent 协议实现的 Broadcast 的具体实现（目前spark中只有一种实现）。它继承自 Broadcast 类，并提供以下功能：

       TorrentBroadcast 包含以下主要成员变量和方法：

       TorrentBroadcast 通过将广播数据分成小块并使用类似 BitTorrent 的协议进行分布式传输，以提高广播性能和可靠性。它允许在集群中高效地广播大量数据，并减少了驱动程序的负载。

       内部版本广播方法broadcastInternal

       该方法是spark内部版本的广播 - 将只读变量广播到集群，变量将仅发送一次到每个执行器。该方法中使用了broadcastManager对象中的newBroadcast创建广播变量

       broadcastManager初始化和创建广播对象初始化

       BroadcastManager构造函数会调用自身的initialize方法，创建一个TorrentBroadcastFactory实例.对象在实例化时，会自动调用自身的writeBlocks，把数据写入blockManager：

       使用了实现了BroadcastFactory接口的TorrentBroadcastFactory工厂方法。TorrentBroadcastFactory 是一个使用类似 BitTorrent 的协议来进行广播数据分布式传输的广播工厂。

       创建广播变量

       TorrentBroadcastFactory实例通过调用newBroadcast() 方法创建新的 TorrentBroadcast对象即广播变量。 可以参考上文实现类

       源码拓展BroadcastManager对象

       BroadcastManager 是 Spark 中负责管理广播变量的类。它包含以下主要功能：

       此外，BroadcastManager 还包含了一些内部变量，如下：

       总而言之，BroadcastManager 提供了广播变量的管理和操作功能，确保广播变量能够在集群中高效地分发和访问。

       BroadcastFactory接口

       BroadcastFactory 是 Spark 中所有广播实现的接口，用于允许多个广播实现。它定义了以下方法：

       通过实现BroadcastFactory 接口，可以自定义广播实现，并在 SparkContext 中使用相应的广播工厂来实例化广播变量。

       TorrentBroadcastFactory

       TorrentBroadcastFactory 是一个使用类似 BitTorrent 的协议来进行广播数据分布式传输的广播工厂。它实现了 BroadcastFactory 接口，并提供以下功能：

       TorrentBroadcastFactory 主要用于支持使用 BitTorrent-like 协议进行分布式传输的广播操作，以提高广播数据在集群中的传输效率和可靠性。

       BitTorrent 协议

       BitTorrent 是一种流行的文件分享协议，它使用了一种名为 "块链" 的技术。块链技术通常用于比特币等加密货币，但在 BitTorrent 中，它用于分发大型文件。

       BitTorrent 的工作原理

       初始化: 当一个用户想要下载一个文件时，他首先创建一个 "种子" 文件，这个文件包含该文件的所有块的哈希列表。 查找: 下载者使用 BitTorrent 客户端软件查找其他下载者，并请求他们分享文件块。 交换: 下载者与其他下载者交换文件块。每个下载者不仅下载文件，还同时通过上传已下载的块来帮助其他下载者。 完整性: 每个块都有一个哈希值，用于验证块的完整性。如果某个块的哈希值不匹配，则该块被认为是无效的，需要重新下载。

       块链技术

       BitTorrent 使用块链来确保每个块的完整性。每个块都包含前一个块的哈希值，这使得整个文件的所有块形成了一个链。如果某个块被修改或损坏，它的哈希值将不再匹配，BitTorrent 客户端将自动从其他下载者那里请求一个新的块。

       安全性

       BitTorrent 协议不使用加密，这意味着在交换文件块时，你的数据可能被第三方监听。为了提高安全性，你可以使用一个加密的 BitTorrent 客户端，如 BitTorrent Secure。

       总结

       BitTorrent 协议是一种高效的文件分享协议，它使用块链技术来保证文件块的完整性和安全性。然而，由于其不加密的特点，它可能不适合传输敏感信息。

Spark源码解析2-YarnCluster模式启动

       YARN 模式运行机制主要体现在Yarn Cluster 模式和Yarn Client 模式上。在Yarn Cluster模式下，SparkSubmit、ApplicationMaster 和 CoarseGrainedExecutorBackend 是独立的进程，而Driver 是独立的线程；Executor 和 YarnClusterApplication 是对象。在Yarn Client模式下，SparkSubmit、ApplicationMaster 和 YarnCoarseGrainedExecutorBackend 也是独立的进程，而Executor和Driver是对象。

       在源码中，SparkSubmit阶段首先执行Spark提交命令，底层执行的是开启SparkSubmit进程的命令。代码中，SparkSubmit从main()开始，根据运行模式获取后续要反射调用的类名赋给元组中的ChildMainClass。如果是Yarn Cluster模式，则为YarnClusterApplication；如果是Yarn Client模式，则为主类用户自定义的类。接下来，获取ChildMainClass后，通过反射调用main方法的过程，反射获取类然后通过构造器获取一个示例并多态为SparkApplication，再调用它的start方法。随后调用YarnClusterApplication的start方法。在YarnClient中，new一个Client对象，其中包含了yarnClient = YarnClient.createYarnClient属性，这是Yarn在SparkSubmit中的客户端，yarnClient在第行初始化和开始，即连接Yarn集群或RM。之后就可以通过这个客户端与Yarn的RM进行通信和提交应用，即调用run方法。

       ApplicationMaster阶段主要涉及开启一个Driver新线程、AM向RM注册、AM向RM申请资源并处理、封装ExecutorBackend启动命令以及AM向NM通信提交命令由NM启动ExecutorBackend。在ApplicationMaster进程中，首先开启Driver线程，开始运行用户自定义代码，创建Spark程序入口SparkContext，接着创建RDD，生成job，划分阶段提交Task等操作。

       在申请资源之前，AM主线程创建了Driver的终端引用，作为参数传入createAllocator()，因为Executor启动后需要向Driver反向注册，所以启动过程必须封装Driver的EndpointRef。AM主线程向RM申请获取可用资源Container，并处理这些资源。ExecutorBackend阶段尚未完成，后续内容待补充。

Spark-Submit 源码剖析

       直奔主题吧：

       常规Spark提交任务脚本如下：

       其中几个关键的参数：

       再看下cluster.conf配置参数，如下：

       spark-submit提交一个job到spark集群中，大致的经历三个过程：

       代码总Main入口如下：

       Main支持两种模式CLI：SparkSubmit；SparkClass

       首先是checkArgument做参数校验

       而sparksubmit则是通过buildCommand来创建

       buildCommand核心是AbstractCommandBuilder类

       继续往下剥洋葱AbstractCommandBuilder如下：

       定义Spark命令创建的方法一个抽象类,SparkSubmitCommandBuilder刚好是实现类如下

       SparkSubmit种类可以分为以上6种。SparkSubmitCommandBuilder有两个构造方法有参数和无参数：

       有参数中根据参数传入拆分三种方式，然后通过OptionParser解析Args，构造参数创建对象后核心方法是通过buildCommand，而buildCommand又是通过buildSparkSubmitCommand来生成具体提交。

       buildSparkSubmitCommand会返回List的命令集合，分为两个部分去创建此List,

       第一个如下加入Driver_memory参数

       第二个是通过buildSparkSubmitArgs方法构建的具体参数是MASTER，DEPLOY_MODE，FILES，CLASS等等，这些就和我们上面截图中是对应上的。是通过OptionParser方式获取到。

       那么到这里的话buildCommand就生成了一个完成sparksubmit参数的命令List

       而生成命令之后执行的任务开启点在org.apache.spark.deploy.SparkSubmit.scala

       继续往下剥洋葱SparkSubmit.scala代码入口如下：

       SparkSubmit，kill，request都支持，后两个方法知识支持standalone和Mesos集群方式下。dosubmit作为函数入口，其中第一步是初始化LOG，然后初始化解析参数涉及到类

       SparkSubmitArguments作为参数初始化类，继承SparkSubmitArgumentsParser类

       其中env是测试用的,参数解析如下，parse方法继承了SparkSubmitArgumentsParser解析函数查找 args 中设置的--选项和值并解析为 name 和 value ，如 --master yarn-client 会被解析为值为 --master 的 name 和值为 yarn-client 的 value 。

       这之后调用SparkSubmitArguments#handle(MASTER, "yarn-client")进行处理。

       这个函数也很简单，根据参数 opt 及 value，设置各个成员的值。接上例，parse 中调用 handle("--master", "yarn-client")后，在 handle 函数中，master 成员将被赋值为 yarn-client。

       回到SparkSubmit.scala通过SparkSubmitArguments生成了args，然后调用action来匹配动作是submit,kill,request_status,print_version。

       直接看submit的action，doRunMain执行入口

       其中prepareSubmitEnvironment初始化环境变量该方法返回一个四元 Tuple ，分别表示子进程参数、子进程 classpath 列表、系统属性 map 、子进程 main 方法。完成了提交环境的准备工作之后，接下来就将启动子进程。

       runMain则是执行入口，入参则是执行参数SparkSubmitArguments

       Main执行非常的简单：几个核心步骤

       先是打印一串日志（可忽略），然后是创建了loader是把依赖包jar全部导入到项目中

       然后是MainClass的生成，异常处理是ClassNotFoundException和NoClassDeffoundError

       再者是生成Application，根据MainClass生成APP,最后调用start执行

       具体执行是SparkApplication.scala，那么继续往下剥~

       仔细阅读下SparkApplication还是挺深的，所以打算另外写篇继续深入研读~

Spark RDD中cache和persist的区别

       é€šè¿‡è§‚察RDD.scala源代码即可知道cache和persist的区别：

       def persist(newLevel: StorageLevel): this.type = {

       ã€€ã€€if (storageLevel != StorageLevel.NONE && newLevel != storageLevel) {

       ã€€ã€€ã€€ã€€throw new UnsupportedOperationException( "Cannot change storage level of an RDD after it was already assigned a level")

       ã€€ã€€}

       ã€€ã€€sc.persistRDD(this)

       ã€€ã€€sc.cleaner.foreach(_.registerRDDForCleanup(this))

       ã€€ã€€storageLevel = newLevel

       ã€€ã€€this

       }

       /** Persist this RDD with the default storage level (`MEMORY_ONLY`). */

       def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

       /** Persist this RDD with the default storage level (`MEMORY_ONLY`). */

       def cache(): this.type = persist()

       å¯çŸ¥ï¼š

       1）RDD的cache()方法其实调用的就是persist方法，缓存策略均为MEMORY_ONLY；

       2）可以通过persist方法手工设定StorageLevel来满足工程需要的存储级别；

       3）cache或者persist并不是action；

       é™„：cache和persist都可以用unpersist来取消