1.SPARK-38864 - Spark支持unpivot源码分析
2.Spark-Submit 源码剖析
3.Spark ML系列RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析
4.spark sql源码系列 | json_tuple一定比 get_json_object更高效吗?
5.源码解析Spark中的码编码Parquet高性能向量化读
SPARK-38864 - Spark支持unpivot源码分析
unpivot是数据库系统中用于列转行的内置函数,如SQL SERVER,码编码 Oracle等。以数据集tb1为例,码编码每个数字代表某个人在某个学科的码编码成绩。若要将此表扩展为三元组,码编码可使用union实现。码编码下程序 picker 源码但随列数增加,码编码SQL语句变长。码编码许多SQL引擎提供内置函数unpivot简化此过程。码编码unpivot使用时需指定保留列、码编码进行转行的码编码列、新列名及值列名。码编码
SPARK从SPARK-版本开始支持DataSet的码编码unpivot函数,逐步扩展至pyspark与SQL。码编码在Dataset API中,码编码ids为要保留的Column数组,Column类提供了从String构造Column的隐式转换,方便使用。利用此API,可通过unpivot函数将数据集转换为所需的三元组。values表示转行列,jforum源码下载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-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的netbeans项目源码命令集合,分为两个部分去创建此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,设置各个成员的RPC实现源码值。接上例,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 ML系列RandomForestClassifier RandomForestClassificationModel随机森林原理示例源码分析
Spark ML中的随机森林分类器(RandomForestClassifier)是一个集成学习方法的分类模型。通过使用多个决策树,它进行自助采样与特征随机选择来构建预测模型。其优势在于能够高效处理大量高维数据,对缺失值和噪声具有鲁棒性,并能评估特征重要性,同时训练过程可并行执行提高速度。参数设置如决策树数量、深度和特征选择策略直接影响模型性能和泛化能力,需根据具体问题和数据集调优以获得最佳效果。
RandomForestClassifier用于Spark ML分类任务,封装在特定类中,支持数据处理与模型训练过程的关键方法。可调整参数优化模型表现,例如特征选择与决策树设置。模型通过构建包含数据转换与训练的Pipeline流程实现自动训练。
以下为基本示例代码:
1. 加载数据集并构建特征向量和标签索引。
2. 将数据集划分为训练集与测试集。
3. 创建RandomForestClassifier实例,并设定关键参数。
4. 构建Pipeline并训练模型。
5. 对测试集进行预测,并评估模型性能,常用指标如多分类准确率。
代码示例中包含实现RandomForestClassifier类的构造与基本用法,如类成员、常量声明和模型对象定义等。此部分源码用于构造随机森林模型的抽象概念与实现基础。
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中的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文件的特定优化,或者其在数据处理流程中的其他改进,但具体原因需要进一步深入分析才能揭示。