1.Spark源码解析2-YarnCluster模式启动
2.spark sql源码系列 | json_tuple一定比 get_json_object更高效吗？
3.spark sql源码系列 | with as 语句真的源译会把查询的数据存内存嘛？
4.SPARK-38864 - Spark支持unpivot源码分析
5.Spark ML系列RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析
6.Spark-Submit 源码剖析

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 sql源码系列 | json_tuple一定比 get_json_object更高效吗？

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

spark sql源码系列 | with as 语句真的会把查询的数据存内存嘛？

       在探讨 Spark SQL 中 with...as 语句是否真的会把查询的数据存入内存之前，我们需要理清几个关键点。首先，网上诸多博客常常提及 with...as 语句会将数据存放于内存中，来提升性能。那么，实际情况究竟如何呢？

       让我们以 hive-sql 的视角来解答这一问题。在 hive 中，有一个名为 `hive.optimize.cte.materialize.threshold` 的参数。默认情况下，其值为 -1，猜骰子源码搭建代表关闭。当值大于 0 时（如设置为 2），with...as 语句生成的表将在被引用次数达到设定值后物化，从而确保 with...as 语句仅执行一次，进而提高效率。

       接下来，我们通过具体测试来验证上述结论。在不调整该参数的情况下，执行计划显示 test 表被读取了两次。此时，我们将参数调整为 `set hive.optimize.cte.materialize.threshold=1`，执行计划显示了 test 表被物化的情况，表明查询结果已被缓存。

       转而观察 Spark SQL 端，我们并未发现相关优化参数。Spark 对 with...as 的操作相对较少，在源码层面，通过获取元数据时所做的参数判断（如阈值与 cte 引用次数），我们可以发现 Spark 在这个逻辑上并未提供明确的优化机制，来专门针对 with...as 语句进行高效管理。

       综上所述，通过与 hive-sql 的对比以及深入源码分析，我们得出了 with...as 语句在 Spark SQL 中是否把数据存入内存的结论，答案并不是绝对的。关键在于是否通过参数调整来物化结果，以及 Spark 在自身框架层面并未提供特定优化策略来针对 with...as 语句进行内存管理。因此，正确使用 with...as 语句并结合具体业务场景，灵活调整优化参数策略，dnf写飞机源码是实现性能提升的关键。

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

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 中设置的--选项和值并解析为 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中的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文件的特定优化，或者其在数据处理流程中的其他改进，但具体原因需要进一步深入分析才能揭示。