Carbondata 2.2.0 集成 Spark3.X - 介绍

个人介绍

开源之夏的小伙伴们，大家好呀。我是王家仪，目前在 Umass 攻读计算机工程硕士同时也是 CarbonData 开源社区的学生成员之一。非常有幸能够参与 CarbonData 2.2.0 的开发，成为一名光荣的 contributor，这次我将把历时三个月的开发经历汇总整理成文章分享给大家，与希望大家能够多多关注 CarbonData 开源社区，为国产开源项目贡献自己的力量~（文末有链接）

项目介绍

首先很高兴地宣布，通过和社区工程师以及其他小伙伴的不断努力，我们已经成功推出了最新的 CarbonData 2.2.0 (Apache CarbonData 2.2.0 Release) 。在这次版本更新中，我们完成了多达48个与性能相关的 JIRA Ticket，同时修复了之前出现的bug，并在此基础上推出了新特性，如：

对复杂列中添加、删除、重命名等操作的支持
支持 Spark 3.1
支持 Presto 二级索引
提升 CDC 性能
提升了本地排序分区加载和压缩的性能
增强了地理空间查询特性
改进了表状态和元数据的写入

这次的开源之夏项目，我参与的是其中 CarbonData 集成支持 Spark 3.1 的部分。

CarbonData 介绍

CarbonData是由华为带头开发的大数据开源项目，2017年正式成为 Apache 软件基金会（ASF）的开源顶级项目。目前已在20+企业生产环境上部署和使应用，企业数据规模达到万亿。针对当前大数据领域分析场景需求各异而导致的存储冗余问题，业务驱动下的数据分析灵活性要求越来越高，CarbonData提供了一种新的融合数据存储方案，以一份数据同时支持多种应用场景，并通过多级索引、字典编码、预聚合、动态Partition、准实时数据查询、列存等特性提升了IO扫描和计算性能，实现百亿数据级秒级响应。

同时，CarbonData也是一种将数据源与Spark集成的高性能分析引擎，这也是为什么在 CarbonData 自身版本更新中对 Spark 的支持显得至关重要。CarbonData作为Spark内部数据源运行，不需要额外启动集群节点中的其他进程，CarbonData Engine在Spark Executor进程之中运行，运行架构如下：

CarbonData 特性

SQL功能：CarbonData与Spark SQL完全兼容，支持所有可以直接在Spark SQL上运行的SQL查询操作。
简单的Table数据集定义：CarbonData支持易于使用的DDL(数据定义语言)语句来定义和创建数据集。CarbonData DDL十分灵活、易于使用，并且足够强大，可以定义复杂类型的Table。

便捷的数据管理：CarbonData为数据加载和维护提供多种数据管理功能。CarbonData支持加载历史数据以及增量加载新数据。加载的数据可以基于加载时间进行删除，也可以撤销特定的数据加载操作。
CarbonData文件格式是HDFS中的列式存储格式。该格式具有许多新型列存储文件的特性，例如，分割表和数据压缩。CarbonData具有以下独有的特点：

伴随索引的数据存储：由于在查询中设置了过滤器，可以显著加快查询性能，减少I/O扫描次数和CPU资源占用。CarbonData索引由多个级别的索引组成，处理框架可以利用这个索引来减少需要安排和处理的任务，也可以通过在任务扫描中以更精细的单元（称为blocklet）进行skip扫描来代替对整个文件的扫描。
可选择的数据编码：通过支持高效的数据压缩，可基于压缩/编码数据进行查询，在将结果返回给用户之前，才将编码转化为实际数据，这被称为“延迟物化”。

支持一种数据格式应用于多种用例场景：例如，交互式OLAP-style查询，顺序访问（big scan），随机访问（narrow scan）

CarbonData 关键技术和优势

快速查询响应：高性能查询是CarbonData关键技术的优势之一。CarbonData查询速度大约是Spark SQL查询的10倍。CarbonData使用的专用数据格式围绕高性能查询进行设计，其中包括多种索引技术和多次的Push down优化，从而对TB级数据查询进行最快响应。
高效率数据压缩：CarbonData使用轻量级压缩和重量级压缩的组合压缩算法压缩数据，可以减少60%~80%数据存储空间，很大程度上节省硬件存储成本。

Spark 3.X 介绍

Spark 3 在之前版本基础上，包含 3,400 多个补丁，是开源社区巨大贡献的结晶，带来了 Python 和 SQL 功能的重大进步，并专注于探索和生产的易用性，其相关的主要新特性有：

通过自适应查询执行、动态分区修剪和其他优化，TPC-DS 的性能比 Spark 2.4 提高了 2 倍
ANSI SQL 合规性
pandas API 的重大改进，包括 Python 类型提示和额外的 Pandas UDF
更好的 Python 错误处理，简化 PySpark 异常
用于结构化流媒体的新 UI
调用 R 语言的用户定义函数速度提升高达 40倍
解决了超过 3,400 个 JIRA Tickets

CarbonData 2.2.0

由于篇幅限制，此处仅介绍两个性能相关新特性（更多资料参见官方wiki）：

1. 自适应查询（AQE, Adaptive Query Execution）

新的自适应查询执行(AQE) 框架通过在运行时生成更好的执行计划来提高性能并简化调优，即使初始计划由于缺少/不准确的数据统计和错误估计的成本而未达到最佳状态——简单来说，我们让机器机器越来越聪明，学会自动优化，新入手的工程师再也不用担心调优经验不够辣~

由于存储和计算分离，数据到达可能是不可预测的。基于所有这些原因，与传统系统相比，运行时自适应性变得更加重要。因此，版本引入了三个主要的自适应优化：

动态聚合分区（Dynamically coalescing shuffle partitions）
动态切换链接策略（Dynamically switching join strategies）
动态优化倾斜（Dynamically optimizing skew joins）

基于 3TB TPC-DS 基准测试，与没有 AQE 的情况相比，具有 AQE 的 CarbonData 可以为两个查询产生超过 1.5 倍的性能加速，对于另外 37 个查询产生超过 1.1 倍的加速。

【注】在 CarbonData 2.2.0 中 AQE 是默认关闭，可以在配置中选择开启。

2. 动态分区修剪（DDP, Dynamic Partition Pruning）

在之前的版本中，引擎的优化器只会对数据查询进行静态分区优化（Static Partition Pruning），如图：

可是实际生产场景中的查询往往更加复杂，在如图所示的场景中，静态分区优化就会几乎失效。

现在，当优化器在编译时无法识别它可以跳过的分区时，将应用动态分区修剪。

从上面可以看出，拥有动态分区裁剪，引擎能够在运行的时候先对 Countries 表的相应字段进行了一次过滤，然后再和 Sales 表进行 Join，可想而知这个性能一般都会有提升的，特别是在 Countries 表有很多无用的数据时性能提升会非常大的。在 TPC-DS 基准测试中，102 个查询中有 60 个显示出 2 倍到 18 倍的显着加速。