前言

一个偶然的机会，从某Spark微信群知道了CarbonData，从断断续续地去了解，到测试 1.2 版本，再到实际应用 1.3 版本的流式入库，也一年有余，在这期间，得到了 CarbonData 社区的陈亮，李昆，蔡强等大牛的鼎力支持，自己也从认识CarbonData 到应用 CarbonData，再到参与社区的转变，感谢他们！

要把CarbonData用得好，姿势必须正确:)，优化步骤就必不可少，熟悉源码的必要性就不言而喻了，因此准备再进一步研究学习CarbonData源码，同时把学习中的一些点滴记录下来。目前，暂定推出四篇博客，从 Create Table，Load Data（DataFrame.write），Select Data，及结合流式入库等四个方面来浅析下 CarbonData 源码。

言归正传哈。

简介

CarbonData 是首个由中国公司发起并捐献给 Apache 基金会的开源项目，于2017年4月正式成为 Apache 顶级项目，由华为开源并支持 Hadoop 的高性能列式存储文件格式，其目的是提供一种统一的数据存储方案，以一份数据同时支持大数据分析的多种应用场景，All In One，并通过多级索引、字典编码、列式存储等特性提升 I/O 扫描和计算性能，实现百亿数据级秒级响应。CarbonData 里程碑版1.3于2018年2月正式发布，该版本包含了集成 Spark 2.2.1，支持预聚合，流式准实时入库，支持标准Hive分区等几个重要特性，而即将发布的1.4版本更是包含了支持 Lucene Index 加强文本检索能力，支持 Bloom Filter Index 等重要特性，在 All In One 的道路上又迈进了一步。

目前 CarbonData 与 Spark，Presto, Hive 等框架做了集成，其中与Spark的集成最深入，提供了基于索引、预聚合、全局字典等更多查询优化手段来提升性能，也提供了数据更新、删除、增量入库等数据管理能力，可以说是在 Spark 开源框架上针对数据仓库类应用的增强。一个计算框架方面的 All In One，一个存储格式方面的 All In One（其实 CarbonData 的功能已经远远超越了数据格式的范畴了），两者结合碰出的火花着实十分吸引人，这也是当初我会想要使用CarbonData 并深入了解它的最根本原因。

此系列就是基于 CarbonData 里程碑版1.3的源码进行浅析。

Create Table浅析

建表语句兼容 SparkSQL，只是使用 'carbondata' 这个DataSource，并扩展了一些属性来描述 Global Dictionary Columns，Sort Columns，Sort Scope 等：

建表流程

如果想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号：iteblog_hadoop

下面对建表流程进行详细的分析。

CarbonHelperSqlAstBuilder.createCarbonTable：

• 生成TableIdentifier；
• 对不支持的建表语句进行验证，比如不支持Temp View，SKEWED BY，CLUSTERED BY及EXTERNAL TABLE；
• 获取column列表，如果有分区字段，则进行合并àfields；
• 如果是streaming表，暂不支持分区表；

CarbonSpark2SqlParser.prepareTableModel：

该方法就是对所有列进行dimension及measure的划分

• 对所有列fields按建表时的顺序进行编号；
• 调用extractDimAndMsrFields方法开始划分dimension和measure列：

1. sortcolumns目前暂不支持如下类型："array", "struct","double", "float", "decimal"，希望后续有更多的人来参与实现；
2. 如果在DICTIONARY_INCLUDE中有定义的列，则加到dimFields；
3. 为TIMESTAMP类型且不存在于dictIncludeCols中的列，则同时加到noDictionaryDims和dimFields；
4. 类型如果是"string", "array","struct", "timestamp", "date", "char"，则加到dimFields，但如果是string类型，则同时需要加入到noDictionaryDims；
5. 如果在SORT_COLUMNS中的列，则同时加入到noDictionaryDims和dimFields；
6. 其他则加入到msrFields；
7. 如果没有定义sort columns，则把dimFields中（除"array", "struct"类型外）的列都作为sortKeyDims；

• 调用extractNoInvertedIndexColumns获取NO_INVERTED_INDEX的列信息；
• 调用getPartitionInfo获取分区信息：支持HASH，RANGE，LIST三种分区类型（CarbonData社区用户主要用的是Hive标准分区，HASH，RANGE，LIST三种分区当前为Alpha特性）；
• 检验TABLE_BLOCKSIZE，只支持1-2048MB的范围；
• 检验TableLevelCompaction属性；
• 对dimFields中列进行重新排序，把复杂（Array，Struct）类型的数据排到最后，普通类型放在前面；

TableNewProcessor(tableModel).process：

• 对TableModel中的sortKeyDims、dimCols、msrCols进行UUID和Encoding设置，生成对应的ColumnSchema：

1. 对于sortKeyDims：出现在noDictionaryDims中的列，都不具有Encoding.DICTIONARY；Date和Timestamp（不在noDictionaryDims中）类型则具有Encoding.DIRECT_DICTIONARY；
2. 对于dimCols：不在sortKeyDims中的列都增加Encoding.DICTIONARY；
3. 对于msrCols：不具有任何Encoding；

• 扫描allColumns，如果列在TableModel.noInvertedIdxCols中或者TableModel.msrCols中，则为NO INVERTED INDEX 列，否则就具有Encoding.INVERTED_INDEX（此部分逻辑可以进一步优化，后续会提交PR到社区，应该是SORT COLUMNS列都默认具有INVERTED INDEX，除非在NO_INVERTED_INDEX属性中特别指定的列，不过该问题不会影响实际的写数据，写数据时的判断逻辑是正确的）；
• 生成val tableInfo = new TableInfo()，val tableSchema = new TableSchema()，tableInfo.setFactTable(tableSchema)；
• 至此，列的划分结束，allColumns中的列是按照先sort column列，非sort column的dimensions列、complex data type 列、measures列的顺序排序；

CarbonCreateTableCommand.processMetadata：

• 生成CarbonTable carbonTable =CarbonTable.buildFromTableInfo(tableInfo)；
• 调用CarbonUtil.convertToMultiGsonStrings，用Gson把TableInfo转为json string，并分割为几个部分，中间会加上一些carbonSchemaPartsNo及carbonSchemaN的字符串，每个部分长度为4000；
• 拼凑实际的Create Table语句，指定DataSource：'USING org.apache.spark.sql.CarbonSource'，并把上一步骤分割的字符串作为建表的options加入到SQL中；

>


如果想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号：iteblog_hadoop
• CarbonSource.updateCatalogTableWithCarbonSchema：把上步骤中建表语句中的carbonSchemaPartsNo和carbonSchemaN属性读取出来拼凑为TableInfo的json string，然后反序列化为TableInfo实例，调用CarbonFileMetastore.saveToDisk(tableInfo, properties("tablePath"))方法，使用ThriftWriter写到Metadata/schema文件中；然后把建表语句中的carbonSchemaPartsNo和carbonSchemaN属性去除；
• CreateDataSourceTableCommand.run：调用sessionState.catalog.createTable(newTable,ignoreIfExists = false)执行建表语句；