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则是执行入口，入参则是htpc hdmi 源码执行参数SparkSubmitArguments

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

       示例源码broadcast方法

       功能：将只读变量广播到集群,返回一个Broadcast对象以在分布式函数中进行读取变量将仅发送一次到每个执行器，同时调用了内部的森林ce源码方法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 ML系列RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析

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

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

       以下为基本示例代码：

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

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

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

        4. 构建Pipeline并训练模型。

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

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

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

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

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

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

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

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

spark sql源码系列 | json_tuple一定比 get_json_object更高效吗？

       对比json_tuple和get_json_object，网上普遍认为json_tuple效率更高。理由是json_tuple仅需解析一次json数据，而get_json_object需多次解析。实际操作中，get_json_object在解析json字符串到jsonObject阶段仅执行一次，而非多次解析。从执行计划角度看，get_json_object更为简洁，而json_tuple涉及udtf函数，其执行计划更为繁重。功能多样性上，get_json_object支持更丰富的路径处理，如正则匹配、嵌套、多层取值等，而json_tuple仅能解析第一层key。在实际使用时，无需盲从效率结论，根据具体需求选择。确保json数据不过长过大，无论使用哪种方法，效率都不会理想。正确理解并合理运用这些函数，对于优化查询性能至关重要。

Spark体系结构的主要功能和组件

       Spark已经成为全球主要行业中功能强大且需求量最大的大数据框架，其可访问性和强大功能使其能够处理大数据挑战。拥有超过,名成员的良好用户基础，以及多人对代码所做的贡献，它已成为阿里巴巴、亚马逊、eBay、雅虎、腾讯、百度等主流企业的首选框架。InMobi数据科学与市场高级副总裁Rajiv Bhat表示：“Spark使开发机器学习模型的时间从六到七个月缩短至每天约四个模型。”Spark作为开源框架，已成为Apache Software Foundation运行最频繁的项目，目前在大数据处理领域是市场领导者。

       Spark与Hadoop之间的主要区别在于，Hadoop基于处理已存储一段时间的数据块的概念，而Spark则用于实时处理。Hadoop在年是大数据领域的突破性技术，但直到年Spark引入时才如此。Spark的主要销售主张是实时速度，因为它比Hadoop的MapReduce框架快倍。Spark功能包括实时数据处理的开放源代码计算集群框架，提供对具有内置并行性和容错性的整个集群进行编程的接口，其核心建立在Hadoop的MapReduce框架上并扩展到更多计算类型。

       Spark体系结构基于两个主要的抽象，包括主/从体系结构，具有一个Master和多个Slave/Worker。单个Java进程由驱动程序和执行程序运行，用户可以在不同机器上运行它们，以适应垂直集群、混合机器配置或在同一水平的Spark集群。Spark体系结构包括驱动程序、执行程序和集群管理器的角色。驱动程序是Spark应用程序的中心点，负责将用户代码转换为实际的Spark作业。执行者主要负责执行任务，而集群管理器提供了不同的调度功能集，以分配和取消分配各种物理资源，如客户端Spark作业、CPU内存等。

       Spark应用程序的运行时架构涉及客户端提交的Spark应用程序代码被转换为逻辑DAG（有向无环图），进行各种优化，如对转换进行流水线处理，然后转换为具有一组阶段的物理执行计划。物理执行计划由各种小型物理执行单元组成，这些任务组合在一起并发送到Spark集群。驱动程序与集群管理器进行交互，进行资源协商，集群管理器在从属节点上启动执行程序。驱动程序根据数据放置将任务发送给集群管理器，执行程序在执行前向驱动程序注册，驱动程序在应用程序运行时进行监视。当驱动程序main（）方法退出或stop（）方法退出时，它将终止所有执行程序并将其从集群管理器中释放。

       因此，Spark架构因其易用性、可访问性以及处理大数据任务的能力，最终在Hadoop中占主导地位。它在许多行业中广泛应用，将Hadoop MapReduce提升至全新水平，在数据处理方面几乎没有任何改组。内存中的数据存储和实时数据处理提高了系统的效率倍，惰性评估有助于速度。关注微信公众号“海牛大数据”（ID：hainiudashuju），加入实战技术论坛，参与大数据技术交流社区。

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- - 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的开发原理与内在机制。