欢迎来到皮皮网网首页

【比特币网络源码解析】【肇源码头图片高清】【github源码怎么打包jar】spark源码阅读环境

来源:java nio 源码 时间:2024-11-26 19:32:38

1.Spark源码解析2-YarnCluster模式启动
2.SPARK-38864 - Spark支持unpivot源码分析
3.如何在Mac使用Intellij idea搭建远程Hadoop开发环境
4.Spark-Submit 源码剖析
5.源码解析Spark中的码阅Parquet高性能向量化读

spark源码阅读环境

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,github源码怎么打包jar并处理这些资源。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的开发原理与内在机制。

如何在Mac使用Intellij idea搭建远程Hadoop开发环境

       (1)准备工作

       1)

       安装JDK

       6或者JDK

       7

       2)

       安装scala

       2..x

       (注意版本)

       2)下载Intellij

       IDEA最新版(本文以IntelliJ

       IDEA

       Community

       Edition

       .1.1为例说明,不同版本,界面布局可能不同)

       3)将下载的Intellij

       IDEA解压后,安装scala插件,流程如下:

       依次选择“Configure”–>

       “Plugins”–>

       “Browse

       repositories”,输入scala,然后安装即可

       (2)搭建Spark源码阅读环境(需要联网)

       一种方法是直接依次选择“import

       project”–>

       选择spark所在目录

       –>

       “SBT”,之后intellij会自动识别SBT文件,并下载依赖的外部jar包,整个流程用时非常长,取决于机器的网络环境(不建议在windows

       下操作,可能遇到各种问题),一般需花费几十分钟到几个小时。注意,下载过程会用到git,因此应该事先安装了git。

       第二种方法是首先在linux操作系统上生成intellij项目文件,然后在intellij

       IDEA中直接通过“Open

       Project”打开项目即可。在linux上生成intellij项目文件的方法(需要安装git,不需要安装scala,sbt会自动下载)是:在

       spark源代码根目录下,输入sbt/sbt

       gen-idea

       注:如果你在windows下阅读源代码,建议先在linux下生成项目文件,然后导入到windows中的intellij

       IDEA中。

       (3)搭建Spark开发环境

       在intellij

       IDEA中创建scala

       project,并依次选择“File”–>

       “project

       structure”

       –>

       “Libraries”,选择“+”,将spark-hadoop

       对应的包导入,比如导入spark-assembly_2.-0.9.0-incubating-hadoop2.2.0.jar(只需导入该jar

       包,其他不需要),如果IDE没有识别scala

       库,则需要以同样方式将scala库导入。之后开发scala程序即可:

       编写完scala程序后,可以直接在intellij中,以local模式运行,方法如下:

       点击“Run”–>

       “Run

       Configurations”,在弹出的框中对应栏中填写“local”,表示将该参数传递给main函数,如下图所示,之后点击“Run”–>

       “Run”运行程序即可。

       如果想把程序打成jar包,通过命令行的形式运行在spark

       集群中,可以按照以下步骤操作:

       依次选择“File”–>

       “Project

       Structure”

       –>

       “Artifact”,选择“+”–>

       “Jar”

       –>

       “From

       Modules

       with

       dependencies”,选择main函数,并在弹出框中选择输出jar位置,并选择“OK”。

       最后依次选择“Build”–>

       “Build

       Artifact”编译生成jar包。

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