皮皮网

1.【Spark ML系列】RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析
2.Spark-Submit 源码剖析
3.spark原理系列 broadcast广播原理优缺点示例源码权威讲解
4.SPARK-38864 - Spark支持unpivot源码分析

【Spark ML系列】RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析

       Spark ML中的码修随机森林分类器（RandomForestClassifier）是一个集成学习方法的分类模型。通过使用多个决策树，码修它进行自助采样与特征随机选择来构建预测模型。码修其优势在于能够高效处理大量高维数据，码修对缺失值和噪声具有鲁棒性，码修并能评估特征重要性，码修低位启动源码指标同时训练过程可并行执行提高速度。码修参数设置如决策树数量、码修深度和特征选择策略直接影响模型性能和泛化能力，码修需根据具体问题和数据集调优以获得最佳效果。码修

       RandomForestClassifier用于Spark ML分类任务，码修封装在特定类中，码修支持数据处理与模型训练过程的码修关键方法。可调整参数优化模型表现，码修例如特征选择与决策树设置。码修模型通过构建包含数据转换与训练的Pipeline流程实现自动训练。

       以下为基本示例代码：

        1. 加载数据集并构建特征向量和标签索引。

        2. 将数据集划分为训练集与测试集。

        3. 创建RandomForestClassifier实例，并设定关键参数。

        4. 构建Pipeline并训练模型。

        5. 对测试集进行预测，并评估模型性能，常用指标如多分类准确率。

       代码示例中包含实现RandomForestClassifier类的构造与基本用法，如类成员、远程访问手机源码常量声明和模型对象定义等。此部分源码用于构造随机森林模型的抽象概念与实现基础。

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 中设置的c 免费源码下载--选项和值并解析为 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还是6221mg源码挺深的，所以打算另外写篇继续深入研读~

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

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

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

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

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

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

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

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

       总之，广播变量适用于需要在多个任务之间共享只读数据，并且能够提供更高效的数据访问和减少网络传输开销的情况。通过使用广播变量，可以提高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- - 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的开发原理与内在机制。