1.Linux下spark安装指南快速实现大数据处理linux安装spark
2.Spark源码解析2-YarnCluster模式启动
3.spark sql源码系列 | json_tuple一定比 get_json_object更高效吗?
4.Spark-Submit 源码剖析
5.源码解析Spark中的码下Parquet高性能向量化读
6.SparkSQL源码分析-05-SparkSQL的join处理
Linux下spark安装指南快速实现大数据处理linux安装spark
Linux下Spark安装指南,快速实现大数据处理
在Big Data领域,码下Apache Spark可谓是码下一种强大的数据处理框架,它把大数据处理变得更加容易、码下高效。码下本文主要介绍如何在Linux系统下安装和使用Spark,码下.net文件下载源码以便高效的码下处理大数据。
首先,码下在安装Spark之前,码下确保在Linux系统上已经安装了JDK(Java Development Kit),码下当然也可以使用其他语言,码下但是码下相比其他语言,在Java的码下环境下,可以让Spark的码下体验更好。
其次,码下下载Spark的源码,例如从Apache官方网站上下载:http://spark.apache.org,下载Spark的最新版本。将下载之后的压缩文件解压到Linux系统目录,例如/Usr/local目录下:
tar -xvf spark-latest.tar.gz
接下来,将Spark安装为 Linux系统服务,这样可以随时开启和关闭Spark;
在终端中输入以下命令,激活Spark Service:
sudo systemctl enable /usr/local/spark/sbin/start-spark.service
最后,可以利用Spark Shell命令,来查看Spark是否安装成功:
./bin/spark-shell
如果可以看到spark对象,那么表明软件安装成功。
通过以上步骤,可以快速在Linux系统上安装Spark,使用它来处理大数据。分割数源码安装完成后,可以使用定义好的Spark应用程序,构建MapReduce应用程序,实现海量数据统计运算,以及算法分析等。借助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,论文源码软件并处理这些资源。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阶段仅执行一次,而非多次解析。从执行计划角度看,get_json_object更为简洁,而json_tuple涉及udtf函数,其执行计划更为繁重。功能多样性上,get_json_object支持更丰富的路径处理,如正则匹配、嵌套、多层取值等,而json_tuple仅能解析第一层key。在实际使用时,无需盲从效率结论,根据具体需求选择。确保json数据不过长过大,无论使用哪种方法,效率都不会理想。正确理解并合理运用这些函数,对于优化查询性能至关重要。扫雷扫尾源码
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文件的特定优化,或者其在数据处理流程中的其他改进,但具体原因需要进一步深入分析才能揭示。
SparkSQL源码分析--SparkSQL的join处理
SparkSQL的join处理策略多样,针对不同场景各有优劣。首先,map join适用于小表广播至worker节点,提升性能,但大表可能导致OOM。shuffle hash join则对大表进行分区和排序,效率高但内存密集。默认策略通过sort merge join,对大表进行分区排序,避免内存问题,但需预先排序。
当常规策略不可用时,会考虑等值或不等值join的广播nested loop join,适用于特定条件的right或left outer join。笛卡尔积join在无指定key时使用,仅限inner join。
SparkPlan中的Join子节点与策略紧密相关,如在等值连接时,根据hint选择Broadcast hash join、Shuffle sort merge join或shuffle hash join。没有hint时,依据表大小、join类型和排序情况自动选择。
非等值连接时,hint会引导使用broadcast nested loop join或Cartesian product join,无hint时则依据表大小和连接类型来决定。
在特殊情况下,如NotInSubquery,仍可能选择Broadcast hash join。总的来说,SparkSQL的join策略灵活多变,旨在根据具体场景提供最优的执行效率和资源利用率。
Spark源码分析——yarn-cluster模式下Application提交源码实现
Spark源码深入解析:yarn-cluster模式下Application提交的详细流程 Spark客户端在yarn-cluster模式下的核心入口是org.apache.spark.deploy.yarn.Client,这个客户端主要职责是向ResourceManager提交并监控Application的运行。以下是对submit源码的深入剖析: 1. 客户端入口与主要方法: Client的main方法首先创建Client实例并执行run()方法,run()方法是核心操作。 2. submitApplication()核心实现: run()方法中的关键步骤是submitApplication(),它包含了以下内容:初始化Yarn客户端,通过org.apache.hadoop.yarn.client.api.YarnClient实现,向RM申请新应用,生成YarnClientApplication。
检查用户提交的资源(如executorMemory、driverMemory)是否合法,确保不超过单个container的最大资源。
创建ContainerLaunchContext,包括上传依赖资源到HDFS,设置Java执行命令(包含ApplicationMaster入口类)和环境变量。
设置application的详细信息,如名称、队列、资源需求等,然后提交至RM启动ApplicationMaster进程。
3. 资源验证与container创建: 验证用户设置的资源是否满足container限制,并创建执行环境,包括打包依赖文件到HDFS,构建启动ApplicationMaster的Java命令。 4. 监控与报告: 客户端通过monitorApplication()持续监控应用状态并报告给用户,如:Application report for $appId (state: $state)。 总结来说,yarn-cluster模式下,Client执行的步骤包括:创建Client实例,连接ResourceManager。
提交申请,获取applicationId和最大资源。
检查并确保资源请求合法。
构建ContainerLaunchContext,准备application的运行环境。
设置并提交application信息,启动ApplicationMaster。
持续监控并报告application状态。
这个过程完成后,ApplicationMaster的运行和Driver的控制将作为后续分析的重点。