spark SQL not only SQL

1.SparkSession/Data /Datasets API2.Catalyst Optimization & Tungsten Execution3.DataSource Connectors/  Spark Core(RDD API)

优化尽可能的发生晚些，因为spark SQL，可以通过函数和库优化
整体的优化使用库和sql/data

RUN EXPLAIN planInterpret  plantune plan

https://dbricks.co/2rR8vAr

optimizer：
使用启发式和代价重写查询计划

        column pruning：列裁剪， outer join elimination：消除outer join        Predicate push down：谓词下推， constraint propagation：约束传播（broadcast）        constant floding：常量累加： join reordering： join重排序        .....    spark.sql.autoBroadcastJoin.threshold    keep the statistics updated     broadcastJoin Hintmemory manager:    跟踪内存的使用，有效的分配内存在task和算子    code generator： 编译物理计划到优化后的java代码Tungsten Engine： 高效的二进制数据格式和数据结构对cpu和内存的高效。调整spark.sql.codegen.hugeMethodLimit去避免较大的方法（>8k)， 因为这不能被JIT编译器所编译。

spark分为计算和存储:

完整的数据流：    外部存储给spark 喂数据    spark处理数据如果spark处理数据很快，数据源就可能称为瓶颈。更高效的去读取柱状的向量化数据更高效的使用jvm生成simd 说明指定的文件系统可以完成跳过不必要的数据和预shuffle，可以通过不必要的shuffle和IO来加速查询选择支持向量化读取的数据源（parquet，orc)基于文件的数据源，尽可能的创建分区，桶。

Spark 2.3.0支持ORC Vectorized矢量化源码分析

在Spark2.3.0的release文档中，提到ORC Vectored带来的性能提升：

提高scan吞吐2-5倍；
开启条件：spark.sql.orc.impl=native；

ORC 文件类型

当然该ISSUE的提出还是有些背景的(https://issues.apache.org/jira/browse/SPARK-16060)，ORC文件格式本身是Hortonworks提出的针对Hive查询的一种列式存储方案，ORC是在一定程度上扩展了RCFile，是对RCFile的优化。有别于Facebook的RCFile类型，ORC有如下优点：

• ORCFile在RCFile基础上引申出来Stripe和Footer等。每个ORC文件首先会被横向切分成多个Stripe，而每个Stripe内部以列存储，所有的列存储在一个文件中，而且每个stripe默认的大小是250MB，相对于RCFile默认的行组大小是4MB，所以比RCFile更高效；
• ORCFile扩展了RCFile的压缩，除了Run-length（游程编码），引入了字典编码和Bit编码；
• ORCFile保存了文件更多的元信息；

其存储格式如下：

IndexData中保存了该stripe上数据的位置信息，总行数等信息RowData以stream的形式保存了数据的具体信息Stripe Footer中包含该stripe的统计结果，包括Max，Min，count等信息IndexDataRowDataStripeFooter...FileFooter中包含该表的统计结果，以及各个Stripe的位置信息Post s中存储该表的行数，压缩参数，压缩大小，列等信息

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12

ORC Vectored使用场景

spark.sql.orc.impl=native的判断

其中spark.sql.orc.impl有native和hive两种选择，如果针对orc类型选择hive格式直接调用org.apache.spark.sql.hive.orc.OrcFileFormat类实现类的加载，而如果为native则会基于org.apache.spark.sql.execution.datasources.orc.OrcFileFormat类进行加载。

 /** Given a provider name, look up the data source class definition. */  def lookupDataSource(provider: String, conf: SQLConf): Class[_] = {    val provider1 = backwardCompatibilityMap.getOrElse(provider, provider) match {      case name if name.equalsIgnoreCase("orc") &&          conf.getConf(SQLConf.ORC_IMPLEMENTATION) == "native" =>        classOf[OrcFileFormat].getCanonicalName      case name if name.equalsIgnoreCase("orc") &&          conf.getConf(SQLConf.ORC_IMPLEMENTATION) == "hive" =>        "org.apache.spark.sql.hive.orc.OrcFileFormat"      case name => name    }    ...  }  

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14

org.apache.spark.sql.execution.datasources.orc.OrcFileFormat

什么时候支持ORC Vectored?

  override def supportBatch(sparkSession: SparkSession, schema: StructType): Boolean = {    val conf = sparkSession.sessionState.conf    conf.orcVectorizedReaderEnabled && conf.wholeStageEnabled &&      schema.length <= conf.wholeStageMaxNumFields &&      schema.forall(_.dataType.isInstanceOf[AtomicType])  }

• 1
• 2
• 3
• 4
• 5
• 6

需要满足以下条件：
* 开启spark.sql.orc.enableVectorizedReader: 默认true;
* 开启spark.sql.codegen.wholeStage: 默认true并且其scheme的长度不大于wholeStageMaxNumFields（默认100列）;
* [关键]所有列数据类型需要为AtomicType类型的；

AtomicType类型，可根据定义查看：

/** * An internal type used to represent everything that is not null, UDTs, arrays, structs, and maps. */protected[sql] abstract class AtomicType extends DataType {  private[sql] type InternalType  private[sql] val tag: TypeTag[InternalType]  private[sql] val ordering: Ordering[InternalType]}

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8

AtomicType代表了非 null/UDTs/arrays/structs/maps类型。所以如果所含列中如果包含null/UDTs/arrays/structs/maps类型，依然无法收到该ISSUE的便利。

ORC Vectored实现

OrcColumnarBatchReader的使用

在OrcFileFormat.buildReaderWithPartitionValues中：

    if (enableVectorizedReader) {          val batchReader = new OrcColumnarBatchReader(            enableOffHeapColumnVector && taskContext.isDefined, copyToSpark)          // SPARK-23399 Register a task completion listener first to call `close()` in all cases.          // There is a possibility that `initialize` and `initBatch` hit some errors (like OOM)          // after opening a file.          val iter = new RecordReaderIterator(batchReader)          Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => iter.close()))          // 调用initialize函数          batchReader.initialize(fileSplit, taskAttemptContext)          // 调用initBatch          batchReader.initBatch(            reader.getSchema,            requestedColIds,            requiredSchema.fields,            partitionSchema,            file.partitionValues)          // 生成iter          iter.asInstanceOf[Iterator[InternalRow]]    } else {          val orcRecordReader = new OrcInputFormat[OrcStruct]            .createRecordReader(fileSplit, taskAttemptContext)          val iter = new RecordReaderIterator[OrcStruct](orcRecordReader)          Option(TaskContext.get()).foreach(_.addTaskCompletionListener(_ => iter.close()))          val fullSchema = requiredSchema.toAttributes ++ partitionSchema.toAttributes          val unsafeProjection = GenerateUnsafeProjection.generate(fullSchema, fullSchema)          val deserializer = new OrcDeserializer(dataSchema, requiredSchema, requestedColIds)          if (partitionSchema.length == 0) {            iter.map(value => unsafeProjection(deserializer.deserialize(value)))          } else {            val joinedRow = new JoinedRow()            iter.map(value =>              unsafeProjection(joinedRow(deserializer.deserialize(value), file.partitionValues)))          }    }

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34
• 35
• 36
• 37
• 38

OrcColumnarBatchReader的实现

initialize()： 初始化OrcFile Reader及Hadoop环境配置；
initBatch(): 初始化batch变量和columnarBatch变量(其中batch为ORC Reader矢量化每次读取的结果存储变量，columnarBatch为codegen转换为Spark定义类型存储变量 )；
nextBatch(): 迭代器，其核心还是调用ORC自定义的vectored函数，需要根据类型转换Spark定义type；

单元测试

参考： org.apache.spark.sql.hive.orc.OrcReadBenchmark

结果分析：
针对数字、String类型测试

Native ORC Vectorized > Native ORC Vectorized with copy > Native ORC MR > Hive built-in ORC

针对分区、不分区测试

Partition性能远远>不分区性能

参考：

• 
  
• http://spark.apache.org/releases/spark-release-2-3-0.html
• https://www.cnblogs.com/ITtangtang/p/7677912.html
• RCFile和ORCFile: http://blog.csdn.net/u014307117/article/details/52381383