CarbonData实践(一)

前言

CarbonData 拥有不错的明细查询能力，比如简单的where条件过滤，性能大概是Parquet的20倍。数据的聚合分析方面，如果有不错的where过滤，则相当一部分查询也是快于Parquet的，并且拥有更少的Tasks数，这就意味着可以让你的Spark Query Service 有更好的并发能力。

查询和入库的性能，一般而言是都是抵触的，需要有一个较好的权衡。CarbonData 在这块和Parquet 有一定的差距。

环境

Spark 1.6.0 + CarbonData 1.0.0

你可以通过这篇文章的介绍 迅速搭建一个基于CarbonData存储，以Spark为计算引擎的 Rest Service 服务。

数据导入

我们尝试两个规模数据的导入：

1. 1000万数据的导入
2. 12亿数据的导入（原始表24亿）

Spark 版本为 CDH5.7 Spark 1.6.0 ,对应所有的配置参数：

 --conf "spark.sql.shuffle.partitions=30" \
 --master yarn-cluster \
 --num-executors 12 \
 --executor-memory 10G \
 --executor-cores 1 \
 --driver-memory 6G \

导入都是通过SQL完成，类似

insert into table select * from another table

CarbonData有较复杂的存储结构，需要构建字典以及MDK等，所以整个构建过程中，需要两个大阶段：

1. 构建全局字典文件
2. 创建数据文件

所谓全局字典，其实指的是将你的列值用一个数字来存储和表示，这是列式存储的一个常用技巧。

假设你导入的数据规模很大，你需要将高基的列通过DICTIONARY_EXCLUDE 排除掉。否则很可能数据导入过程就OOM，然后executor掉了。所谓高基指的是某列count(distinct)的值。一般而言高于五万十万以上的基，而你的资源又非常有限，那么可以排除掉，不建字典。

在写入数据文件时，需要构建MDK。所谓MDK,其实就是一个多字段的复合索引，按声明顺序把所有字段拼接起来，越靠近左边的越有序，CarbonData的明细查询就靠这个了。还有倒排索引，通过列反向找到行。不过这也就意味着需要有全局排序了，排序一直是个消耗内存和CPU的活，

因为内存放不下，所以一般都要支持外排，所以在资源有限的情况下，你想导入大量的数据，需要注意两个地方：

1. 减少单Executor导入任务的并行度，并行度越高，占用的内存就越大。
2. 减少排序过程中的内存使用

对于第一种，行之有效的参数是：

carbon.number.of.cores.while.loading=2
carbon.number.of.cores.block.sort=1
carbon.merge.sort.reader.thread=1

第二情况则是：

carbon.sort.size=5000
carbon.sort.file.write.buffer.size=5000
carbon.merge.sort.prefetch=false

第一个和第二个参数会使得排序过程产生大量的磁盘文件。第三个参数则使得排序过程不从临时文件预取数据，这块消耗内存很大，如果资源较少，推荐关闭，可以有效缓解任务失败的概率。

还有一些值得注意的地方

1. insert into table A select * from another table B

这里需要注意的是,A和B表字段顺序需要保持一致。否则可能发生字段的错位。

类型也需要匹配，无法转换的字段会为null。比如如果B字段是Date类型，在CarbonData里配置成timestamp 则会变成null从而引起一些诡异问题。

1. Query的一些优化

** Spark 方面的一些调优**

首先 spark.speculation 是一个值得打开的参数，不至于让某个慢的Task导致整个查询过慢。

其次 ** spark.sql.shuffle.partitions ** 设置为你的CPU数的一到三倍是个不错的选择。因为过多的分区数，而你的CPU数又不足的情况下，会让费很多时间。

再有就是CPU充足不妨把 spark.locality.wait=0 设置上，这样可以让你的CPU充分跑起来而不会几个忙着，其他的嫌着。代价也是有的，可能产生更多的网络流量。

** Query语句上的一些调优 **

这几天调试了很多上百行的SQL，非常多和复杂的子查询，Join查询，大体的规则有：

1. join的时候，一定要在on语句后面加上一些过滤条件，减少被join的数据量
2. 把where过滤条件尽量放到子查询里去，因为在子查询外面的where条件没法下沉，导致子查询计算量非常大。
3. 利用好MDK的索引特性，尽量将表左边的字段作为过滤条件

CarbonData的一些调优

有一种情况，就是单条记录非常小，那么一个CarbonData文件哪怕是几十M,那么可能也包含了几百万条记录。CarbonData有非常好的剪枝能力，可以不用去touch 那些不包含数据的block文件,所以这个时候可以让CarbonData文件小一点，经过剪枝后，虽然可能文件会多一些，但是每个文件小，并且能够提升并行度，从而有效的提升查询的效率。这个通过创建表的过程中设置TABLE_BLOCKSIZE来达到目标。在一个实际案例中，block大小是64M,后面我改成8M后，性能提升十倍左右。CarbonData 在1.0.0里默认是1G。这个值还是看大家row的大小来调整。

第二个情况是Block的数目以及大小分布不均。我们在实际案例发现，某张表有四个Block,其中一个Block是另外3个Block的的四倍大小。这样会发生严重的拖尾问题。第一是在导入数据的过程中，尽量利用多个节点，因为CarbonData生成的Block数和你的节点数是有关系的。其次是你可以设置前面的TABLE_BLOCKSIZE。

Query的优化，还有很多配置方面的参数可以做，比如向量化reader的开启，这块我还没实测，先不多说。