概述
本文介绍Spark SQL增加的Columnar模块代码实现。
首先介绍Columnar内的代码结构和实现,然后介绍在SqlContext里的使用方式。
Columnar
InMemoryColumnarTableScan
实现
InMemoryColumnarTableScan类是SparkPlan LeafNode的实现,即是一个物理执行计划。
private[sql] case class InMemoryColumnarTableScan(attributes: Seq[Attribute], child: SparkPlan)
extends LeafNode {
传入的child是一个SparkPlan(确认了的物理执行计划)和一个属性序列。
行转列并cache的过程如下:
lazy val cachedColumnBuffers = {
val output = child.output
// 遍历每个RDD的partiti on
val cached = child.execute().mapPartitions { iterator =>
// 把属性Seq转换成为ColumnBuilder数组
val columnBuilders = output.map { attribute =>
// 都是基本ColumnBuilder,默认ByteBuffer大小
ColumnBuilder(ColumnType(attribute.dataType).typeId, 0, attribute.name)
}.toArray
var row: Row = null
// RDD每个Partition的Rows,每个Row的所有field信息存到ColumnBuilder里
while (iterator.hasNext) {
row = iterator.next()
var i = 0
while (i < row.length) {
columnBuilders(i).appendFrom(row, i)
i += 1
}
}
Iterator.single(columnBuilders.map(_.build()))
}.cache()
cached.setName(child.toString)
// Force the materialization of the cached RDD.
cached.count()
cached
}
ColumnType类用于表示Column的类型,他的typeId变量用来区分数据类型,生成对应的ColumnBuilder(typeId, initialSize=0, columnName)。ColumnBuilder的生成如下:
def apply(typeId: Int, initialSize: Int = 0, columnName: String = ""): ColumnBuilder = {
val builder = (typeId match {
case INT.typeId => new IntColumnBuilder
case LONG.typeId => new LongColumnBuilder
case FLOAT.typeId => new FloatColumnBuilder
case DOUBLE.typeId => new DoubleColumnBuilder
case BOOLEAN.typeId => new BooleanColumnBuilder
case BYTE.typeId => new ByteColumnBuilder
case SHORT.typeId => new ShortColumnBuilder
case STRING.typeId => new StringColumnBuilder
case BINARY.typeId => new BinaryColumnBuilder
case GENERIC.typeId => new GenericColumnBuilder
}).asInstanceOf[ColumnBuilder]
builder.initialize(initialSize, columnName)
builder
}
他的继承结构如下,主要有三大体系:
这里涉及到的是Basic这个体系,继承结构如下:
BasicColumnBuilder里,initialSize = 0,指使用ByteBuffer的默认大小,即10*1024*104。然后在initialize()方法,会初始化ByteBuffer。
接下来,针对RDD每个partition,
var row: Row = null
while (iterator.hasNext) {
row = iterator.next()
var i = 0
while (i < row.length) {
columnBuilders(i).appendFrom(row, i)
i += 1
}
}
进行了appendFrom操作:
override def appendFrom(row: Row, ordinal: Int) {
val field = columnType.getField(row, ordinal)
buffer = ensureFreeSpace(buffer, columnType.actualSize(field))
columnType.append(field, buffer)
}
用于把一个Row的每一个field,都存到一个ColumnBuilder里。在这里指BasicColumnBuilder这个类,维护了一个自己的ByteBuffer,把row里的各个field信息都存在了buffer里。
最后ColumnBuilders里的每个ColumnBuilder进行build(),即BasicColumnBuilder.build()方法,进行了一次ByteBuffer的rewind()方法。
这个方法的结果是一个RDD集合。由于在结束前调用了.count()方法,所以RDD的计算是被执行了的,返回的是新的RDD。
在Spark SQL里,外部调用cachedColumnBuffers方法只有在uncache table的时候,进行了unpersisit()操作。
下面看execute()方法:
override def execute() = {
cachedColumnBuffers.mapPartitions { iterator =>
// 在RDD partition里,iterator.next()返回的是一个ByteBuffer
// 也就是说,cachedColumnBuffers返回的结果RDD,类型是ByteBuffer
val columnBuffers = iterator.next()
assert(!iterator.hasNext)
new Iterator[Row] {
// 访问每一个ByteBuffer里的列信息
val columnAccessors = columnBuffers.map(ColumnAccessor(_))
val nextRow = new GenericMutableRow(columnAccessors.length)
override def next() = {
var i = 0
// 把column里的信息再转到Row里
while (i < nextRow.length) {
columnAccessors(i).extractTo(nextRow, i)
i += 1
}
nextRow
}
override def hasNext = columnAccessors.head.hasNext
}
}
}
使用
在SqlContext里选择cache table的时候,会使用该类。
其实在cache的时候,首先去catalog里寻找这个table的信息和table的执行计划,然后会进行执行(执行到物理执行计划生成),然后把这个table再放回catalog里维护起来,这个时候的执行计划已经是最终要执行的物理执行计划了。但是此时Columner模块相关的转换等操作都是没有触发的。
真正的触发还是在execute()的时候,同其他SparkPlan的execute()方法触发场景是一样的。
ColumnBuilder 与 ColumnAccessor
一个包装Row的每个field成Column;一个访问column,然后可以转回Row
关于压缩
private[sql] abstract class NativeColumnBuilder[T <: NativeType](
override val columnStats: NativeColumnStats[T],
override val columnType: NativeColumnType[T])
extends BasicColumnBuilder[T, T#JvmType](columnStats, columnType)
with NullableColumnBuilder
with AllCompressionSchemes
with CompressibleColumnBuilder[T]
private[sql] class BooleanColumnBuilder extends NativeColumnBuilder(new BooleanColumnStats, BOOLEAN)
private[sql] class IntColumnBuilder extends NativeColumnBuilder(new IntColumnStats, INT)
从继承结构看,压缩的builder和Accessor都以trait的方式继承了ColumnBuilder,而子类比如IntColumnBuilder,不但继承了BaseColumnBuilder,同时也具备压缩处理能力。
具体压缩处理可以参考CompressibleColumnBuilder类里的实现。
是否压缩会做一次判断,压缩比在0.8以下才执行压缩。
在build()的时候实施压缩,并且按照以下结构存在bytebuffer内。
* .--------------------------- Column type ID (4 bytes)
* | .----------------------- Null count N (4 bytes)
* | | .------------------- Null positions (4 x N bytes, empty if null count is zero)
* | | | .------------- Compression scheme ID (4 bytes)
* | | | | .--------- Compressed non-null elements
* V V V V V
* +---+---+-----+---+---------+
* | | | ... | | ... ... |
* +---+---+-----+---+---------+
* \-----------/ \-----------/
* header body
CompressionScheme子类是不同的压缩实现
都是scala实现的,未借助第三方库。不同的实现,指定了支持的column data类型。在build()的时候,会比较每种压缩,选择压缩率最小的(若仍大于0.8就不压缩了)。
这里的估算能力,在子类实现里,好像是由gatherCompressibilityStats方法实现的。
SqlContext
分析SqlContext内目前cache和uncache table的实现细节与Columnar的关系。
Cache Table
/** Caches the specified table in-memory. */
def cacheTable(tableName: String): Unit = {
// 得到的是一个logicalPlan
val currentTable = catalog.lookupRelation(None, tableName)
// 物理执行计划生成之后交给InMemoryColumnarTableScan
val asInMemoryRelation =
InMemoryColumnarTableScan(currentTable.output, executePlan(currentTable).executedPlan)
// SparkLogicalPlan接受的Plan必须是已经确定plan好的SparkPlan
catalog.registerTable(None, tableName, SparkLogicalPlan(asInMemoryRelation))
}
从上面那段代码可以看到,cache之前,需要先把本次cache的table的物理执行计划生成出来。上述的currentTable其实是一个logicalPlan,来自catalog的lookupRelation。
最后注册表的时候,涉及到的SparkLogicalPlan类是LogicalPlan的实现类(但是本身其实是一个SparkPlan),它接受的是SparkPlan,并且是已经确定Plan好了的逻辑执行计划,目前接受两类:ExistingRdd和InMemoryColumnarTableScan。
在cache这个过程里,InMemoryColumnarTableScan并没有执行,但是生成了以InMemoryColumnarTableScan为物理执行计划的SparkLogicalPlan,并存成table的plan。
Uncache Table
在这一步,除了删除catalog里的table信息之外,还调用了InMemoryColumnarTableScan的cacheColumnBuffers方法,得到RDD集合,并进行了unpersist()操作。cacheColumnBuffers方法具体见Columner内,主要做了把RDD每个partition里的ROW的每个Field存到了ColumnBuilder内。
全文完 :)
分享到:
相关推荐
1.2.1:内存列存储(In-Memory Columnar Storage) 6 1.2.2:字节码生成技术(bytecode generation,即CG) 6 1.2.3:scala代码优化 7 二:sparkSQL运行架构 8 2.1:Tree和Rule 9 2.1.1:Tree 10 2.1.2:Rule 10 2.2...
Chapter 10, Columnstore Indexes, revises the columnar storage and then explores the huge improvements for columnstore indexes in SQLServer 2016: updateable nonclustered columnstore indexes, column...
齐奥·科伦纳尔 即将搬到zio org
藏经阁-The Columnar Era_ Leveraging P.pdf
- Columnar and Grouped Reports - Bi Dashboards and Elements - Synchronizing Groups, Charts, and Sparklines - Chart and Gauge Reports - Creating a Personal Report Card - Creating a Multi-Series Multi-Y...
大数据框构建要求columnar 在单元测试期间使用 zstd 压缩工具。描述这是一个惯用的 kotlin 数据框工具包,用于支持任何规模的数据集集合的数据工程任务。 该工具包的主要重点是在 Dataframe 迭代器上支持类似 Pandas...
藏经阁-The Columnar Era_ Leveraging Parquet, Arrow and Kudu for High-Performance Analytics.pdf
In-Memory Columnar Store for PostgreSQLHorizontal data representationSymbol Day Open High Low Close VolumeAAA 2014-04-27 10.11 10.25 10.08 10.15 125AAB 2014-04-27 40.33 40.50 40.20 40.45 70ABB 2014-...
在FLEX中实现柱状专题图层。在地图中对一些地图要素做数据统计,可以直观获得差异进行比较分析,适用于各种统计数据的直观表示。
Manticore Columnar Library是一个面向列的存储库,旨在在大数据量下以较低的内存占用量提供良好的性能。当与结合使用时,对于那些寻找以下内容的人可能会有所帮助:日志分析,包括丰富的免费文本搜索功能...
HAWQ Hadoop With Query (是一个Hadoop原生大规模并行SQL分析引擎,针对的是分析性应用) Chukwa Apache Chukwa is an open source data collection system for monitoring large distributed systems. Kudu Kudu ...
vc++ 显示音效的柱状频谱模块vc + + shows the spectrum of columnar sound module
该项目提供了一组全面的示例查询(SQL)以及Java代码,以获取和处理与SQL查询匹配的WARC记录。 构建Java工具 mvn package Python,PySpark,Jupyter笔记本 不属于该项目。 请查看 ,以有关如何使用Python和PySpark...
cstore_fdw 实现了 PostgreSQL 数据库的柱状存储,用于对批量加载的数据进行分析的场景。 该扩展使用了 Optimized Row Columnar (ORC) 格式的数据存储布局。ORC 提升 Facebook 开发的 RCFile 格式,带来如下好处: ...
密码学应用 应用程序使用列关键字密码对文本...要求 IDE,例如 (可选) 安装 下载存储库或使用克隆对其进行克隆 ...如何使用 转到src文件夹 $ cd Cryptography-Application/src 运行Main.java $ javac Main.java ...结果
使用Minibase的列式数据库实现 该项目涉及使用关系型DBMS(即Minibase)作为主干来开发成熟的柱状DBMS。 在此列式DBMS的基础上构建了几个查询运算符,它们在下面进行了描述: 支持的索引类型: ...
柱状换位Java 一个简单的 Java GUI 程序,它接受用户的输入(从 a 到 z ),无论字符串是小写还是大写,或者字符串之间是否有空格,并使用密钥( String )进行加密和...Columnar-Transposition-Java 在 MT 许可下发布
tion of queries and columnar data representation. Snel is designed for fast on-line analytics by leveraging the LLVM compiler infrastructure. It also has custom special methods like resolving ...
过去几年开源社区已经发布了很多工具来完善大数据分析的生态系统,这些工具涵盖了数据分析的各个层面,比如列式存储格式(Parquet/ORC)、内存计算层(Drill、Spark、Impala和Storm)以及强大的API接口(Python和R语言)。...
Clickhouse是一个用于联机分析处理(OLAP)的列式数据库管理系统(columnar DBMS)。 传统数据库在数据大小比较小,索引大小适合内存,数据缓存命中率足够高的情形下能正常提供服务。但残酷的是,这种理想情形最终会...