关于CarbonData+Spark SQL的一些应用实践和调优经验分享

   <p>大数据时代，中大型企业数据的爆发式增长，几乎每天都能产生约 100GB 到 10TB 的数据。而企业数据分系统构建与扩张，导致不同应用场景下大数据冗余严重。行业亟需一个高效、统一的融合数仓，从海量数据中快速获取有效信息，从而洞察机遇、规避风险。</p>    <p>在这样的现状下，CarbonData 诞生了，作为首个由中国贡献给Apache社区的顶级开源项目，CarbonData 提供了一种新的融合数据存储方案，以一份数据同时支持多种大数据应用场景，并通过丰富的索引技术、字典编码、列存等特性提升了 IO 扫描和计算性能，实现了PB数据级的秒级响应。</p>    <p>为了帮助开发者深入了解并学习这项大数据开源技术，华为 CarbonData PMC 陈亮牵头，携手技术社区的核心开发者及合作伙伴，举办了一场Apache CarbonData+Spark 主题的技术交流会，就 CarbonData+Spark 的重要特性和使用介绍，做了全面而细致的分享，本文简单整理了其中的部分精彩内容，同时，作为本次活动的承办方，InfoQ整理上传了所有讲师的演讲PPT+演讲视频，感兴趣的同学可以免费获取现场完整资料 。</p>    <h2>Spark SQL的发展史概述 <a href="http://www.open-open.com/doc/view/7239fa57c80a47189382718a1d0e43de" rel="nofollow,noindex">（讲师PPT下载）</a></h2>    <p>来自美国Databricks公司的范文臣首先讲述了Spark SQL的发展史，范文臣同时也是Apache Spark PMC member，主导 Spark SQL 一些主要功能的设计和研发，定期审计项目代码质量等。现场，他将Spark SQL过去的发展分为四个阶段：</p>    <p><img src="https://simg.open-open.com/show/c83abf8553e4902585036fea5f63af99.png"></p>    <ul>     <li><strong>2009年，著名的Spark框架诞生。</strong> 它是一个围绕速度、易用性和复杂分析构建的大数据处理框架，由伯克AMP实验室创建。相比于当时流行的Hadoop，Spark提供了更高效的MapReduce模型，减少数据落地，也降低了编程难度。</li>     <li>2011年，Spark团队将Hive的底层物理执行模块从Hadoop切换成Shark，启动了Shark项目。 然而，由于Hive自身的代码复杂性以及和Hadoop MapReduce的耦合，Shark的开发举步维艰，进展缓慢。</li>     <li>2014年，Spark团队舍弃Shark，重新建立了一套完整的查询框架Catalyst。 Catalyst利用了函数式风格的不可变特性，使Query Plan不可变，优化器通过遍历优化策略生成新的 Query Plan。这样优化规则之间的影响更容易理解，提升了代码的可读性和可维护性，也方便了新特性的开发。下图为Spark SQL控制框架：</li>    </ul>    <p><img src="https://simg.open-open.com/show/0e72cce14bae1869a456fbe974f2122d.png"></p>    <ul>     <li>2015年，Spark团队提出了钨丝计划，通过建立Tungsten格式、后端优化、代码生成等手段，将Spark的查询性能和执行速度提升到了一个新的台阶。</li>     <li>2017年，持续探索中……</li>    </ul>    <p>那么，沿着查询性能这条路，Spark的未来还会有哪些优化方向？范文臣在最后的演讲中总结到：Spark的愿景是管理各种不同性质数据集和数据源的大数据处理的需求。Spark这样一个角色，只关注于计算层，快速查询处理是Spark唯一的衡量标准，也是未来不变的发展方向。也因此，在之后的Spark2.3里面，在计算框架下如何更快的和储存系统桥接、Spark代码生成都是未来着重关注的方向。</p>    <h2>CarbonData应用实践+2.0新技术规划介绍  <a href="/misc/goto?guid=4959754466315065683" rel="nofollow,noindex">（讲师PPT下载）</a></h2>    <p>CarbonData诞生之初是希望以一份数据去满足企业各种各样的场景需求，包括详单过滤和海量数仓以及数据集式操作等。那么，开发者该如何正确使用CarbonData技术？华为CarbonData总设计师李昆结合实际案例，详细讲解了CarbonData应用实践+2.0新技术规划。</p>    <h3>CarbonData大数据生态</h3>    <p>Carbondata在数据查询方面选择和Spark结合，据李昆现场介绍，Carbondata+Spark可以打造一个相对于传统系统来说，更好的交互分析体验，目前Carbondata和Spark1.5、1.6、2.1，Hive，Presto都做了集成，未来还将对Spark2.2做支持；在接口方面，Carbondata提供SQL接口，也支持Spark DataFrame API；在操作方面，支持查询、数据管理如批量入库、更新、删除等操作。</p>    <p><img src="https://simg.open-open.com/show/310adb6ed2a248634b864cbd12e13659.png"></p>    <p>随后，李昆就CarbonData索引建立、CarbonData表格与物理存储、SQL引擎对接、数据管理过程等技术内容做了详细介绍。由于篇幅限制，本文不在此介绍，感兴趣的读者可以下载讲师PPT对CarbonData的存储原理进行深入了解。</p>    <h3>成功案例介绍</h3>    <p>随后，李昆通过电信详单分析场景的举例介绍，详细说明CarbonData如何以一份数据支持多种应用场景的。李昆表示，在电信跟金融领域经常需要明细数据分析，优化之前，老的系统需要用Impala和Hbase两个系统，建立4个二级索引才可以完成业务需要的性能。这其中，Impala用来做报表输出，Hbase做关键维度查询。这两个系统有各自存在不足：Impala没有办法很好的扩展，HBase要做很多二级索引，无法使用yarn统一资源管理，只能是一个个集群单独维护。</p>    <p><img src="https://simg.open-open.com/show/a7b9c949002351ba892b56b2454fb628.png"></p>    <p>用Carbondata+Spark数据优化后，可以解决既要点查又要处理报表的情况。下图是一个从2000亿到1万亿的性能测试数据，Q1是过滤查询，Q2也是过滤查询，Q1跟Q2数据查询因为用了Carbondata索引，需要扫描的数据不会增长很多，数据量增长5倍，查询时间增长不到1倍。第三个查询是full scan查询，主要考察的是spark和carbon的可扩展性，测试过程中发现扩展性是非常线性的，scalability很好。</p>    <p><img src="https://simg.open-open.com/show/891828c6bca6986ad860970fa2f66067.png"></p>    <h3>CarbonData2.0未来规划</h3>    <p>现在，Carbondata的主要特性是对多场景的支持，不过在大数据时代，更多的场景正扑面而来。包括SQL分析、时间序列分析、位置轨迹、文本检索、图查询和机器学习等。这就需要Carbondata2.0在各领域的应用上有更多的准备。包括：</p>    <ul>     <li>入库方面，需要考虑实时事件的流式入库、历史事件的批量入库等；</li>     <li>存储方面分三层，一层是界面，每一个领域有自己的术语，会针对领域常见操作做些SQL上的扩展；二是数据组织层，对不同领域做不同的分区、索引和预处理等，以便于它更高效地存储领域数据；三是存储格式层，Carbondata目前是列存，为了支撑更多查询和分析，数据格式本身也需要具有扩展能力，比如行存、时序、面向AI的格式等；</li>    </ul>    <h2>Spark 2.2 核心特性CBO介绍 <a href="/misc/goto?guid=4959754466392812324" rel="nofollow,noindex">（讲师PPT下载）</a></h2>    <p>在Spark SQL的Catalyst优化器中，许多基于规则的优化技术已经实现，但优化器本身仍然有很大的改进空间。Spark 2.2在Spark SQL引擎内添加了一个基于成本的优化器框架，此框架通过可靠的统计和精确的估算，能够在以下领域做出好的判定：选择散列连接操作的正确构建端，选择正确的连接算法，调整连接的顺序等等，这个基于成本的优化器就是CBO。据华为研究工程师王振华介绍，CBO的目标是希望优化器能够自动为用户选择最优的执行计划，要达到这件事情，需要以下三个步骤：</p>    <p>第一步收集、推断和传播关于源/中间数据的表/列统计信息。用户运行 ANALYZE TABLE 命令会收集表格信息比如表的行数、大小，列的统计信息比如最大值、最小值、不同值个数等，并将这些信息存储到metastore里面。</p>    <p><img src="https://simg.open-open.com/show/88d5ce7b07cbb87c183996f22be81604.png"></p>    <p>第二步Cardinality Estimation，根据收集到的信息，计算每个操作符的成本，包括输出行数、输出大小等。如做filter时写一个过滤条件，给定的条件会基于条件里面涉及列的统计信息，估算过滤条件执行完了以后，Operator有多少数据。</p>    <p><img src="https://simg.open-open.com/show/08cae9fb1b1eb6eba86cc1bb00aaae63.png"></p>    <p>如下图，为一个A小于等于某数字的估算，如果A的value比A的最小值更小，或者是比A的最大值更大，那么过滤率肯定是0或者100%，当落在定义域中间的时候，假设是均匀分布，概率则是A.min到B的区间所占A的定义域的百分比，这个是Filter条件最终的selectivity，有了selectivity，即可再相应的更新filter以后的统计信息。</p>    <p>第三步根据成本计算，选择最优的查询执行计划。通过建造方选择（Build Side Selection）、散列连接实现：广播与洗牌（Hash Join Implementation: Broadcast vs. Shuffle）、多路连接重新排序（Multi-way Join Reorder）、连接成本计算公式（Join Cost Formula）四个方面阐述了最优计划的选择过程。</p>    <p>其中，在多路连接重新排序方法上，采用了动态规划算法。以四表连接为例，首先，将所有项(基本连接节点)放到0级；然后，从第0级的计划中构建所有的两表连接；第三，从以前的层级(单节点和两表连接)中构建出可能的三表连接；最后，构建所有的4路连接，并在其中选出最优的计划。而在构建m-路径连接时，只需保留同一组m项的最佳计划(最优子解决方案)。如，对于A、B、C的三表连接顺序，只保留三个候选计划:(A J B)J C，(A J C)J B和(B J C)J A 当中最优的计划。</p>    <p>Join cost计算方式如下，首先Cost一般来说传统的数据库里是基于CPU和IO，这两个Cost是线性加合。在Spark中，用Cardinality模拟CPU的开销，用size模拟IO的开销。</p>    <p>王振华最后介绍到，华为在2016年7月份开始将CBO贡献给Spark社区，并建立了umbrella ticket - SPARK-16026。截至目前为止，创建了超过40个sub-tasks、提交了50余个pull requests并被合入，同时吸引了十余个社区贡献者的参与。</p>    <p>CBO的第一个版本已经在Spark 2.2中发布，感兴趣的开发者和使用者，如要使用CBO，可以在收集统计信息之后，打开spark.sql.cbo.enable来使用CBO。</p>    <h2>Partition 功能详解+上汽实践分享 <a href="/misc/goto?guid=4959754466477755522" rel="nofollow,noindex">（讲师PPT下载）</a></h2>    <p>CarbonData的partition特性将在Apache CarbonData 1.2.0版本里正式发布，此特性将显著提升大数据查询性能。上汽集团大数据将CarbonData作为平台基础组件，以应对迅猛增长的数据量，那么上汽集团在使用CarbonData过程中遇到了哪些问题？上汽集团大数据平台开发经理曹鲁就CarbonData的partition特性以及上汽集团在CarbonData项目的实践和测试数据做了分享。</p>    <p>曹鲁首先介绍了文件结构，索引生成过程，初次性能测试等主题内容，引出Partition特性带来改变，主要包括两点：1、数据将基于Partition列更为集中存储，查询时可过滤掉大量block，减少spark task数量；2、可以使其他列在排序中更靠前，提升查询性能。</p>    <h3>Partition Table的数据加载及查询过程详解</h3>    <p>随后，曹鲁详细介绍了CarbonData Partition相关的DDL语法，如Create Partition Table、Show Partition等，以及CarbonData Partition Table的数据加载以及查询过程。下图可以很清晰的看到CarbonData Partition的整个数据加载过程。</p>    <p><img src="https://simg.open-open.com/show/9c63045c556525fbb7984a6778c5c827.png"></p>    <p>关于CarbonData Partition Table查询过程，大概分为两个部分：</p>    <ul>     <li>根据SQL中的过滤条件=, <=, <, >, >=, in, not in以及表达式右值确定命中的partitionId</li>     <li><img src="https://simg.open-open.com/show/90ab0b63f12c0db46fb632a9d512162d.png"></li>     <li>如果有其他在排过序的维度列有过滤条件，则在driver端根据B-tree索引获取blocklet  所在的文件名，如没有则获取全部，再根据文件名中的partitionId，筛选得到需要读取的文件，最后再下发spark task进行读取；</li>     <li><img src="https://simg.open-open.com/show/de8c8e5ff66b50116686acfcfffa92b6.png"></li>    </ul>    <p>之后，曹鲁就Partition的新增(add)、拆分(split)及删除(drop)功能的语法和实现过程展开了分析，其中重点区分了Drop Partition但保留数据RangePartition/ListPartition两种Drop Partition类型的不同语法与实现，感兴趣的读者可以下载讲师PPT深入了解。</p>    <h3>上汽在CarbonData项目的实践分享</h3>    <p>在案例分享环节，曹鲁以上汽的数据作为测试数据，分析了CarbonData Partition table和非Partition table条件下的加载性能和查询性能对比。并给出了CarbonData Partition的性能调优建议。本文为大家展示其中的无排序维度列作为过滤条件，有partition列上的范围过滤条件的聚合查询情况的对比结果，如图不难看出，原始查询方式的耗时是添加partition性能查询方式耗时的25倍。</p>    <p><img src="https://simg.open-open.com/show/2bbe69f26331408af1b411851b6a670a.png"></p>    <p>曹鲁给出的CarbonData Partition的性能调优建议：1、 选择最合适的Partition列；2、尽可能的使用Partition列作为过滤条件，例如Partition列为A，开发者根据业务需求在Column B上有筛选条件，但注意到A与B列之间存在某种固定的mapping关系，这时就可以根据B列的过滤条件再新增一个partition列的过滤条件，以提高查询效率。</p>    <h2>现场精彩问答整理</h2>    <p>Q： 客户在使用Spark时不愿意编写代码，更喜欢给他一个页面能能够直接生成SQL，Spark后面会不会更多的偏向于业务人员做一些更易应用的东西出来，比如可以直接出来一个页面？</p>    <p>A：Spark本身不会往这方面走，因为Spark只专注于做计算这层，这个模式一般是另外一个项目，比如有项目zpplin是专门做供应GIU的，可以在zpplin上面调Spark的一些接口，这些会单独立项，而不是在Spark里面做。</p>    <p>Q：刚才提到carbon有一个目标，能够尽量多的支持各种场景，目前我们也做过一些测试，某些特定情况下，不同的场景可能在响应速度和并发性上有比较大的差距，这一点后面有没有改善？ </p>    <p>A：这方面需要跟Spark一起联合做优化，因为Spark是端到端的，从元数据查询到SQL优化到DAG调度执行，有很多中间过程处理会耗时，建议你做一下打点分析，看主要瓶颈是哪一块，同时carbon和spark我们也可以做一些联合优化，相信基于社区的努力后面会有改善。</p>    <p>Q：如果有新的数据添加进来，CarbonData统计信息如何更新？</p>    <p>A：有两种方式，一种是比较简单的，每次数据表更新重新计算增量，这样比较精确但是会比较慢，另外一种方式是增量的更新统计信息，这种方式较前一种可能会稍微复杂一些。</p>    <p>Q：在用Spark写Carbondata Partition的时候，并行比较高，导致每个分区下出现很多小文件，这样有什么好的解决办法？</p>    <p>A：在CarbonData中每一个Block的大小是可以设置的，Blocklet也可以设置的，在load数据的时候，写满一个block的默认大小就会重新再写一个文件，所以可以设置Block大小来解决这个问题。另外定期使用CarbonData的compaction功能也可以合并一些小文件，当然后面我们也会考虑开发merge partition的功能来给用户提供更多选择。</p>    <p> </p>    <p>来自：http://www.infoq.com/cn/news/2017/09/CarbonData-Spark-huawei</p>    <p> </p>