BT

您是否属于早期采用者或者创新人士?InfoQ正在努力为您设计更多新功能。了解更多

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

| 作者 江柳 关注 0 他的粉丝 发布于 2017年9月12日. 估计阅读时间: 1 分钟 | 硅谷人工智能、机器学习、互联网金融、未来移动技术架构 ,尽在QCon上海2017

大数据时代,中大型企业数据的爆发式增长,几乎每天都能产生约 100GB 到 10TB 的数据。而企业数据分系统构建与扩张,导致不同应用场景下大数据冗余严重。行业亟需一个高效、统一的融合数仓,从海量数据中快速获取有效信息,从而洞察机遇、规避风险。

在这样的现状下,CarbonData 诞生了,作为首个由中国贡献给Apache社区的顶级开源项目,CarbonData 提供了一种新的融合数据存储方案,以一份数据同时支持多种大数据应用场景,并通过丰富的索引技术、字典编码、列存等特性提升了 IO 扫描和计算性能,实现了PB数据级的秒级响应。

为了帮助开发者深入了解并学习这项大数据开源技术,华为 CarbonData PMC 陈亮牵头,携手技术社区的核心开发者及合作伙伴,举办了一场Apache CarbonData+Spark 主题的技术交流会,就 CarbonData+Spark 的重要特性和使用介绍,做了全面而细致的分享,本文简单整理了其中的部分精彩内容,同时,作为本次活动的承办方,InfoQ整理上传了所有讲师的演讲PPT,感兴趣的同学可以下载讲师PPT获取完整资料 。

Spark SQL的发展史概述(讲师PPT下载)

来自美国Databricks公司的范文臣首先讲述了Spark SQL的发展史,范文臣同时也是Apache Spark PMC member,主导 Spark SQL 一些主要功能的设计和研发,定期审计项目代码质量等。现场,他将Spark SQL过去的发展分为四个阶段:

  • 2009年,著名的Spark框架诞生。它是一个围绕速度、易用性和复杂分析构建的大数据处理框架,由伯克AMP实验室创建。相比于当时流行的Hadoop,Spark提供了更高效的MapReduce模型,减少数据落地,也降低了编程难度。
  • 2011年,Spark团队将Hive的底层物理执行模块从Hadoop切换成Shark,启动了Shark项目。然而,由于Hive自身的代码复杂性以及和Hadoop MapReduce的耦合,Shark的开发举步维艰,进展缓慢。
  • 2014年,Spark团队舍弃Shark,重新建立了一套完整的查询框架Catalyst。Catalyst利用了函数式风格的不可变特性,使Query Plan不可变,优化器通过遍历优化策略生成新的 Query Plan。这样优化规则之间的影响更容易理解,提升了代码的可读性和可维护性,也方便了新特性的开发。下图为Spark SQL控制框架:

           

  • 2015年,Spark团队提出了钨丝计划,通过建立Tungsten格式、后端优化、代码生成等手段,将Spark的查询性能和执行速度提升到了一个新的台阶。
  • 2017年,持续探索中……

那么,沿着查询性能这条路,Spark的未来还会有哪些优化方向?范文臣在最后的演讲中总结到:Spark的愿景是管理各种不同性质数据集和数据源的大数据处理的需求。Spark这样一个角色,只关注于计算层,快速查询处理是Spark唯一的衡量标准,也是未来不变的发展方向。也因此,在之后的Spark2.3里面,在计算框架下如何更快的和储存系统桥接、Spark代码生成都是未来着重关注的方向。

CarbonData应用实践+2.0新技术规划介绍 (讲师PPT下载)

CarbonData诞生之初是希望以一份数据去满足企业各种各样的场景需求,包括详单过滤和海量数仓以及数据集式操作等。那么,开发者该如何正确使用CarbonData技术?华为CarbonData总设计师李昆结合实际案例,详细讲解了CarbonData应用实践+2.0新技术规划。

CarbonData大数据生态

Carbondata在数据查询方面选择和Spark结合,据李昆现场介绍,Carbondata+Spark可以打造一个相对于传统系统来说,更好的交互分析体验,目前Carbondata和Spark1.5、1.6、2.1,Hive,Presto都做了集成,未来还将对Spark2.2做支持;在接口方面,Carbondata提供SQL接口,也支持Spark DataFrame API;在操作方面,支持查询、数据管理如批量入库、更新、删除等操作。

随后,李昆就CarbonData索引建立、CarbonData表格与物理存储、SQL引擎对接、数据管理过程等技术内容做了详细介绍。由于篇幅限制,本文不在此介绍,感兴趣的读者可以下载讲师PPT对CarbonData的存储原理进行深入了解。

成功案例介绍

随后,李昆通过电信详单分析场景的举例介绍,详细说明CarbonData如何以一份数据支持多种应用场景的。李昆表示,在电信跟金融领域经常需要明细数据分析,优化之前,老的系统需要用Impala和Hbase两个系统,建立4个二级索引才可以完成业务需要的性能。这其中,Impala用来做报表输出,Hbase做关键维度查询。这两个系统有各自存在不足:Impala没有办法很好的扩展,HBase要做很多二级索引,无法使用yarn统一资源管理,只能是一个个集群单独维护。

用Carbondata+Spark数据优化后,可以解决既要点查又要处理报表的情况。下图是一个从2000亿到1万亿的性能测试数据,Q1是过滤查询,Q2也是过滤查询,Q1跟Q2数据查询因为用了Carbondata索引,需要扫描的数据不会增长很多,数据量增长5倍,查询时间增长不到1倍。第三个查询是full scan查询,主要考察的是spark和carbon的可扩展性,测试过程中发现扩展性是非常线性的,scalability很好。

CarbonData2.0未来规划

现在,Carbondata的主要特性是对多场景的支持,不过在大数据时代,更多的场景正扑面而来。包括SQL分析、时间序列分析、位置轨迹、文本检索、图查询和机器学习等。这就需要Carbondata2.0在各领域的应用上有更多的准备。包括:

  • 入库方面,需要考虑实时事件的流式入库、历史事件的批量入库等;
  • 存储方面分三层,一层是界面,每一个领域有自己的术语,会针对领域常见操作做些SQL上的扩展;二是数据组织层,对不同领域做不同的分区、索引和预处理等,以便于它更高效地存储领域数据;三是存储格式层,Carbondata目前是列存,为了支撑更多查询和分析,数据格式本身也需要具有扩展能力,比如行存、时序、面向AI的格式等;

Spark 2.2 核心特性CBO介绍(讲师PPT下载)

在Spark SQL的Catalyst优化器中,许多基于规则的优化技术已经实现,但优化器本身仍然有很大的改进空间。Spark 2.2在Spark SQL引擎内添加了一个基于成本的优化器框架,此框架通过可靠的统计和精确的估算,能够在以下领域做出好的判定:选择散列连接操作的正确构建端,选择正确的连接算法,调整连接的顺序等等,这个基于成本的优化器就是CBO。据华为研究工程师王振华介绍,CBO的目标是希望优化器能够自动为用户选择最优的执行计划,要达到这件事情,需要以下三个步骤:

第一步收集、推断和传播关于源/中间数据的表/列统计信息。用户运行 ANALYZE TABLE 命令会收集表格信息比如表的行数、大小,列的统计信息比如最大值、最小值、不同值个数等,并将这些信息存储到metastore里面。

第二步Cardinality Estimation,根据收集到的信息,计算每个操作符的成本,包括输出行数、输出大小等。如做filter时写一个过滤条件,给定的条件会基于条件里面涉及列的统计信息,估算过滤条件执行完了以后,Operator有多少数据。

如下图,为一个A小于等于某数字的估算,如果A的value比A的最小值更小,或者是比A的最大值更大,那么过滤率肯定是0或者100%,当落在定义域中间的时候,假设是均匀分布,概率则是A.min到B的区间所占A的定义域的百分比,这个是Filter条件最终的selectivity,有了selectivity,即可再相应的更新filter以后的统计信息。

第三步根据成本计算,选择最优的查询执行计划。通过建造方选择(Build Side Selection)、散列连接实现:广播与洗牌(Hash Join Implementation: Broadcast vs. Shuffle)、多路连接重新排序(Multi-way Join Reorder)、连接成本计算公式(Join Cost Formula)四个方面阐述了最优计划的选择过程。

其中,在多路连接重新排序方法上,采用了动态规划算法。以四表连接为例,首先,将所有项(基本连接节点)放到0级;然后,从第0级的计划中构建所有的两表连接;第三,从以前的层级(单节点和两表连接)中构建出可能的三表连接;最后,构建所有的4路连接,并在其中选出最优的计划。而在构建m-路径连接时,只需保留同一组m项的最佳计划(最优子解决方案)。如,对于A、B、C的三表连接顺序,只保留三个候选计划:(A J B)J C,(A J C)J B和(B J C)J A 当中最优的计划。

Join cost计算方式如下,首先Cost一般来说传统的数据库里是基于CPU和IO,这两个Cost是线性加合。在Spark中,用Cardinality模拟CPU的开销,用size模拟IO的开销。

王振华最后介绍到,华为在2016年7月份开始将CBO贡献给Spark社区,并建立了umbrella ticket - SPARK-16026。截至目前为止,创建了超过40个sub-tasks、提交了50余个pull requests并被合入,同时吸引了十余个社区贡献者的参与。

CBO的第一个版本已经在Spark 2.2中发布,感兴趣的开发者和使用者,如要使用CBO,可以在收集统计信息之后,打开spark.sql.cbo.enable来使用CBO。

Partition 功能详解+上汽实践分享(讲师PPT下载)

CarbonData的partition特性将在Apache CarbonData 1.2.0版本里正式发布,此特性将显著提升大数据查询性能。上汽集团大数据将CarbonData作为平台基础组件,以应对迅猛增长的数据量,那么上汽集团在使用CarbonData过程中遇到了哪些问题?上汽集团大数据平台开发经理曹鲁就CarbonData的partition特性以及上汽集团在CarbonData项目的实践和测试数据做了分享。

曹鲁首先介绍了文件结构,索引生成过程,初次性能测试等主题内容,引出Partition特性带来改变,主要包括两点:1、数据将基于Partition列更为集中存储,查询时可过滤掉大量block,减少spark task数量;2、可以使其他列在排序中更靠前,提升查询性能。

Partition Table的数据加载及查询过程详解

随后,曹鲁详细介绍了CarbonData Partition相关的DDL语法,如Create Partition Table、Show Partition等,以及CarbonData Partition Table的数据加载以及查询过程。下图可以很清晰的看到CarbonData Partition的整个数据加载过程。

关于CarbonData Partition Table查询过程,大概分为两个部分:

  • 根据SQL中的过滤条件=, <=, <, >, >=, in, not in以及表达式右值确定命中的partitionId
  • 如果有其他在排过序的维度列有过滤条件,则在driver端根据B-tree索引获取blocklet  所在的文件名,如没有则获取全部,再根据文件名中的partitionId,筛选得到需要读取的文件,最后再下发spark task进行读取;

之后,曹鲁就Partition的新增(add)、拆分(split)及删除(drop)功能的语法和实现过程展开了分析,其中重点区分了Drop Partition但保留数据RangePartition/ListPartition两种Drop Partition类型的不同语法与实现,感兴趣的读者可以下载讲师PPT深入了解。

上汽在CarbonData项目的实践分享

在案例分享环节,曹鲁以上汽的数据作为测试数据,分析了CarbonData Partition table和非Partition table条件下的加载性能和查询性能对比。并给出了CarbonData Partition的性能调优建议。本文为大家展示其中的无排序维度列作为过滤条件,有partition列上的范围过滤条件的聚合查询情况的对比结果,如图不难看出,原始查询方式的耗时是添加partition性能查询方式耗时的25倍。

曹鲁给出的CarbonData Partition的性能调优建议:1、 选择最合适的Partition列;2、尽可能的使用Partition列作为过滤条件,例如Partition列为A,开发者根据业务需求在Column B上有筛选条件,但注意到A与B列之间存在某种固定的mapping关系,这时就可以根据B列的过滤条件再新增一个partition列的过滤条件,以提高查询效率。

现场精彩问答整理

Q: 客户在使用Spark时不愿意编写代码,更喜欢给他一个页面能能够直接生成SQL,Spark后面会不会更多的偏向于业务人员做一些更易应用的东西出来,比如可以直接出来一个页面?

A:Spark本身不会往这方面走,因为Spark只专注于做计算这层,这个模式一般是另外一个项目,比如有项目zpplin是专门做供应GIU的,可以在zpplin上面调Spark的一些接口,这些会单独立项,而不是在Spark里面做。

Q:刚才提到carbon有一个目标,能够尽量多的支持各种场景,目前我们也做过一些测试,某些特定情况下,不同的场景可能在响应速度和并发性上有比较大的差距,这一点后面有没有改善? 

A:这方面需要跟Spark一起联合做优化,因为Spark是端到端的,从元数据查询到SQL优化到DAG调度执行,有很多中间过程处理会耗时,建议你做一下打点分析,看主要瓶颈是哪一块,同时carbon和spark我们也可以做一些联合优化,相信基于社区的努力后面会有改善。

Q:如果有新的数据添加进来,CarbonData统计信息如何更新?

A:有两种方式,一种是比较简单的,每次数据表更新重新计算增量,这样比较精确但是会比较慢,另外一种方式是增量的更新统计信息,这种方式较前一种可能会稍微复杂一些。

Q:在用Spark写Carbondata Partition的时候,并行比较高,导致每个分区下出现很多小文件,这样有什么好的解决办法?

A:在CarbonData中每一个Block的大小是可以设置的,Blocklet也可以设置的,在load数据的时候,写满一个block的默认大小就会重新再写一个文件,所以可以设置Block大小来解决这个问题。另外定期使用CarbonData的compaction功能也可以合并一些小文件,当然后面我们也会考虑开发merge partition的功能来给用户提供更多选择。

评价本文

专业度
风格

您好,朋友!

您需要 注册一个InfoQ账号 或者 才能进行评论。在您完成注册后还需要进行一些设置。

获得来自InfoQ的更多体验。

告诉我们您的想法

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我
社区评论

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

讨论

登陆InfoQ,与你最关心的话题互动。


找回密码....

Follow

关注你最喜爱的话题和作者

快速浏览网站内你所感兴趣话题的精选内容。

Like

内容自由定制

选择想要阅读的主题和喜爱的作者定制自己的新闻源。

Notifications

获取更新

设置通知机制以获取内容更新对您而言是否重要

BT