Spark -orc internal

spark 基本概念介绍

RDD

直观上，RDD可理解为下图所示结构，即RDD包含多个Partition（分区），每个Partition代表一部分数据并位于一个计算节点

RDD本质上是Spark中的一个抽象类

RDD包含以下成员方法或属性：

1、compute方法

提供在计算过程中Partition元素的获取与计算方式

2、partition的列表

每一个partition代表一个并行的最小划分单元；

3、dependencies列表

描述RDD依赖哪些父RDD生成，即RDD的血缘关系；

4、partition的位置列表

定义如何最快速的获取partition的数据，加快计算，这个是可选的，可作为本地化计算的优化选项；

5、partitioner方法

定义如何对数据进行分区

前2个函数是所有RDD必须的，后三个可选，所有子RDD（HadoopRDD、MapPartitionRDD、JdbcRDD等）都要继承并实现其中的方法 。

RDD partition

由于RDD的数据量很大，因此为了计算方便，需要将RDD进行切分并存储在各个节点的分区当中，从而当我们对RDD进行各种计算操作时，实际上是对每个分区中的数据进行并行的操作

也就是一份待处理的原始数据会被按照相应的逻辑切分成多分，每份数据对应RDD的一个Partition，partition的数量决定了task的数量，影响程序的并行度

注意，如果使用hive table，此处RDD 的partition不是 hive table 中表的分区，表的不同分区以不同目录来分开存储，要注意区分。

RDD 算子

RDD算子分类，大致可以分为两类，即：

Transformation：转换算子，这类转换并不触发提交作业，完成作业中间过程处理。

常见的 Transformation算子 ，map，flatmap，reducebykey等

2. Action：行动算子，这类算子会触发SparkContext提交Job作业。

常见的 Action算子，collect，saveAsTextFile，count等

为了方便记忆，可以这么理解，action算子一般是在RDD上计算出来一个结果,把结果返回给driver program或保存在文件系统，返回结果非RDD类型，Transformation一般是传入函数对rdd进行操作，返回也是RDD。

RDD 依赖

RDD 和它依赖的父RDD 的关系有两种不同的类型， 即窄依赖（ narrowdependency）和宽依赖（wide dependency）。

窄依赖是指每个父RDD的一个Partition最多被子RDD的一个Partition所使用，例如map、filter、union等操作都会产生窄依赖；

​ 宽依赖是指一个父RDD的Partition会被多个子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都会产生宽依赖；

​ 需要特别说明的是对join操作有两种情况：如果两个RDD在进行join操作时，一个RDD的partition仅仅和另一个RDD中已知个数的Partition进行join，那么这种类型的join操作就是窄依赖，例如图1中左半部分的join操作(join with inputs co-partitioned)；其它情况的join操作就是宽依赖,例如图1中右半部分的join操作(join with inputs not co-partitioned)，由于是需要父RDD的所有partition进行join的转换，这就涉及到了shuffle，因此这种类型的join操作也是宽依赖。

总结：在这里我们是从父RDD的partition被使用的个数来定义窄依赖和宽依赖，因此可以用一句话概括下：如果父RDD的一个Partition被子RDD的一个Partition所使用就是窄依赖，否则的话就是宽依赖。因为是确定的partition数量的依赖关系，所以RDD之间的依赖关系就是窄依赖；由此我们可以得出一个推论：即窄依赖不仅包含一对一的窄依赖，还包含一对固定个数的窄依赖。

一对固定个数的窄依赖的理解：即子RDD的partition对父RDD依赖的Partition的数量不会随着RDD数据规模的改变而改变；换句话说，无论是有100T的数据量还是1P的数据量，在窄依赖中，子RDD所依赖的父RDD的partition的个数是确定的，而宽依赖是shuffle级别的，数据量越大，那么子RDD所依赖的父RDD的个数就越多，从而子RDD所依赖的父RDD的partition的个数也会变得越来越多。

Application

Term	Meaning
Application	用户编写的Spark应用程序,包括一个Driver和多个executors
Application jar	包含用户程序的Jar包
Driver Program	运行main()函数并创建SparkContext进程
Cluster manager	在集群上获取资源的外部服务，如standalone manager,yarn,Mesos
deploy mode	部署模式，区别在于driver process运行的位置
worker node	集群中可以运行程序代码的节点（机器）
Executor	运行在worker node上执行具体的计算任务，存储数据的进程
Task	被分配到一个Executor上的计算单元
Job	由多个任务组成的并行计算阶段，因RDD的Action产生
Stage	每个Job被分为小的计算任务组，每组称为一个stage
DAGScheduler	根据Job构建基于Stage的DAG，并提交Stage给TaskScheduler
TaskScheduler	将TaskSet提交给worker运行，每个executor运行什么task在此分配

​ 一个Spark程序可以被划分为一个或多个Job，划分的依据是RDD的Action算子，每遇到一个RDD的Action操作就生成一个新的Job， 每个spark Job在具体执行过程中，将其划分为一个或多个stage，划分的依据是RDD间的Dependency关系，当遇到Wide Dependency时因需要进行shuffle操作， 故以此为界划分不同的Stage 。

​ Stage是由Task组组成的并行计算，因此每个stage中可能存在多个Task，这些Task执行相同的程序逻辑，只是它们操作的数据
的一个Partition对应一个Task， Partition的数目可以在创建RDD时指定，也可以通过reparation和coalesce等算子重新进行划分，

将上面的概念串联起来，可以得到下面的运行层次图：

spark sql --orc

spark sql中的RDD

Spark SQL 最终将SQL 语句经过逻辑算子树转换成物理算子树。

在物理算子树中，叶子类型的SparkPlan 节点负责从无到有的创建RDD ，每个非叶子类型的SparkPlan 节点等价于在RDD 上进行一次Transformation ，即通过调用execute（）函数转换成新的RDD ，最终执行collect()操作触发计算，返回结果给用户。

重点分析一下叶子节点：

在Spark SQL 中，LeafExecNode 类型的SparkPlan 负责对初始RDD 的创建。

HiveTableScanExec 会根据Hive数据表存储的HDFS 信息直接生成HadoopRDD；FileSourceScanExec 根据数据表所在的源文件生成FileScanRDD ，在2.3后的spark sql中，使用的就是这种方式从orc文件创建RDD。

spark sql orc 读取

spark sql 中读取orc的流程如下图

下面分析下图中标橘黄色的部分，inputsplit和Vectorized 读取。

spark sql orc inputsplit 源码分析

 /*** Create an RDD for non-bucketed reads.* The bucketed variant of this function is [[createBucketedReadRDD]].** @param readFile a function to read each (part of a) file.* @param selectedPartitions Hive-style partition that are part of the read.* @param fsRelation [[HadoopFsRelation]] associated with the read.*/private def createNonBucketedReadRDD(readFile: (PartitionedFile) => Iterator[InternalRow],selectedPartitions: Array[PartitionDirectory],fsRelation: HadoopFsRelation): RDD[InternalRow] = {// 配置项，默认128Mval defaultMaxSplitBytes =fsRelation.sparkSession.sessionState.conf.filesMaxPartitionBytes// 配置项，默认4Mval openCostInBytes = fsRelation.sparkSession.sessionState.conf.filesOpenCostInBytes// 参数，根据不同场景自动进行选择val defaultParallelism = fsRelation.sparkSession.sparkContext.defaultParallelism// 总大小：分区下所有文件 length+opencostbyteval totalBytes = selectedPartitions.flatMap(_.files.map(_.getLen + openCostInBytes)).sumval bytesPerCore = totalBytes / defaultParallelism// maxSplitBytes 后面会以这个参数为跨度，对input 进行 split// 真实场景下（文件大于128M且数量明显多于core时），maxSplitBytes为128Mval maxSplitBytes = Math.min(defaultMaxSplitBytes, Math.max(openCostInBytes, bytesPerCore))logInfo(s"Planning scan with bin packing, max size: $maxSplitBytes bytes, " +s"open cost is considered as scanning $openCostInBytes bytes.")// split后的files,有多个partitionedFile，注意此处的partitionedFile数量并不等于RDD的partition数        量                                   //分区遍历val splitFiles = selectedPartitions.flatMap { partition =>//文件遍历partition.files.flatMap { file =>val blockLocations = getBlockLocations(file)//文件可分割，orc为可分割if (fsRelation.fileFormat.isSplitable(fsRelation.sparkSession, fsRelation.options, file.getPath)) {//对每个文件进行split，每一个maxSplitBytes为一个partitionedFile，若文件小于                           maxSplitBytes，一个文件一个partitionedFile(0L until file.getLen by maxSplitBytes).map { offset =>val remaining = file.getLen - offsetval size = if (remaining > maxSplitBytes) maxSplitBytes else remainingval hosts = getBlockHosts(blockLocations, offset, size)PartitionedFile(partition.values, file.getPath.toUri.toString, offset, size, hosts)}//文件不可分割，如tgz，一个文件一个partiiiton} else {val hosts = getBlockHosts(blockLocations, 0, file.getLen)Seq(PartitionedFile(partition.values, file.getPath.toUri.toString, 0, file.getLen, hosts))}}}.sortBy(_.length)(implicitly[Ordering[Long]].reverse)val partitions = new ArrayBuffer[FilePartition]val currentFiles = new ArrayBuffer[PartitionedFile]var currentSize = 0L/** Close the current partition and move to the next. */def closePartition(): Unit = {if (currentFiles.nonEmpty) {val newPartition =FilePartition(partitions.size,currentFiles.toArray) // Copy to a new Array.partitions += newPartition}currentFiles.clear()currentSize = 0}// 使用next fit 近似算法将x个partitionedFile 分配入 y个partititon中// Assign files to partitions using "Next Fit Decreasing"splitFiles.foreach { file =>if (currentSize + file.length > maxSplitBytes) {closePartition()}// Add the given file to the current partition.currentSize += file.length + openCostInBytescurrentFiles += file}closePartition()new FileScanRDD(fsRelation.sparkSession, readFile, partitions)}

Next Fit算法简介

• NF 算法是用来解决装箱问题的，装箱问题描述如下： S=（S1,S2,…Sn）,其中0< Si ≤ 1, Si称之为第i个物体的体积（或重量），1≤i≤n,现有n个容积（或载重量）为1 的箱子，要求如何设法将S1,S2,…Sn放入尽可能少的箱中

• 为partitionedFile 分配Partition类似装箱问题，经过inputsplit后，有N个小于maxSplitBytes 的partitionedFile，每个partition的容量为maxSplitBytes，如何将partitionedFile装入更少的partition中

• NF算法的处理方法是始终维持一个当前打开的箱子，对于每一个要装入的物品，检查该物品是否可以放入当前打开的箱子，如果无法装入，则打开一个空箱子，装入该物品，以该箱子作为当前的箱子 。

• spark先将split好的partitionedFile 按照size从大到小排序，然后依次装入到size为maxSplitBytes的partition中，装不下则打开下一个分区，直到所有partitionedFile 都被分配完毕，使用的partition数量就是此RDD partition的数量。

问题

SplitBytes 不是 64m 的整数倍时,orcReader会跨越stripe读取，会不会造成读取的数据不完整?

不会导致读取数据错误，ORC 自身的reader 接口支持按偏移量进行读取。

具体读取方式 后面在详细了解

部分参数解释

Property Name	Default	Meaning
spark.sql.shuffle.partitions	200	Configures the number of partitions to use when shuffling data for joins or aggregations.
spark.default.parallelism	For distributed shuffle operations like reduceByKey and join, the largest number of partitions in a parent RDD. For operations like parallelize with no parent RDDs, it depends on the cluster manager: - Local mode: number of cores on the local machine - Mesos fine grained mode: 8 - Others: total number of cores on all executor nodes or 2, whichever is larger	Default number of partitions in RDDs returned by transformations like join, reduceByKey, and parallelize when not set by user.

看起来它们的定义似乎也很相似，但在实际测试中，

• spark.default.parallelism只有在处理RDD时才会起作用，对Spark SQL的无效, inputsplit 时也用到了此参数。
• spark.sql.shuffle.partitions则是对sparks SQL专用的设置

spark sql orc Vectorized 分析

在Spark2.3.0的release中，支持了ORC Vectored，提到ORC Vectored带来的性能提升：

• 提高scan吞吐2-5倍；

  使用ORC Vector需要满足以下条件

• 开启 spark.sql.orc.impl =native

• 开启 spark.sql.orc.enableVectorizedReader: 默认true;

• 开启spark.sql.codegen.wholeStage: 默认true并且其scheme的长度不大于 wholeStageMaxNumFields（默认100列）;

• [关键]所有列数据类型需要为AtomicType类型的；

​ AtomicType代表了非 null/UDTs/arrays/structs/maps类型

OrcColumnarBatchReader 读取类分析

读取一个file的数据时，orcfileformat(orc的读写入口)会先初始化一个默认orcReader用来做schema验证，初始化Vector等操作，然后调用OrcColumnarBatchReader的initialize()，initBatch()，nextBatch()等方法进行数据读取。

• initialize()： 初始化OrcFile Reader及Hadoop环境配置，会初始化一个带有start,end,sarg 等条件的orcReader

• initBatch(): 初始化batch变量和columnarBatch变量(其中batch为ORC Reader矢量化每次读取的结果存储变量，columnarBatch为codegen转换为Spark定义类型存储变量 )；

  读取时会根据copyTospark参数判断是复制还是引用orc Vector。

• nextBatch(): 迭代器，其核心还是调用ORC自定义的vectored函数，需要根据类型转换Spark定义type；

性能比较
Native ORC Vectorized > Native ORC Vectorized with copy > Native ORC MR > Hive built-in ORC 

Tips

1. 尽量使用 Native ORC Vectorized 读取方式
2. spark.sql.defaultMaxSplitBytes = orc.stripe.size = hdfs.block.size ,设置为64m的整数倍
3. orc.file.size 要大于defaultMaxSplitBytes ，尽量为defaultMaxSplitBytes 的整数倍。
4. spark.sql.shuffle.partitions，要根据集群业务和core数量设置，默认200往往不是最佳

感谢以下文章，借鉴了其中不少篇幅

https://blog.csdn.net/CRISPY_RICE/article/details/79452007

https://segmentfault.com/a/1190000021297920

https://www.cnblogs.com/zlslch/p/5942204.html

https://blog.csdn.net/x_i_y_u_e/article/details/46765093