hive on spark总体设计

http://www.csdn.net/article/2015-04-24/2824545

HIve on spark 总体设计思路,尽可能重用Hive逻辑层面的功能;从省城物理计划开始,提供一整套针对spark的实现,比如SparkCompiler,SparkTask等,这样HIve的查询就可以作为Spark的任务来执行了。一下是几点主要的设计原则。

1、尽可能减少对Hive原有代码的修改,这是和之前的Shark设计思路最大的不同,Shark对HIve的修改动太大以至于无法被Hive社区接受,HIve on spark尽可能减少改动HIve的代码,从而不影响Hive目前对MapReduce和Tez的支持,同时,HIve on spark 保证对现有的MapReduce和Tez模式在功能性能方面不会有任何影响。

2、对于选择Spark的用户,应使其能够自动的获取hive现有和未来新增的功能。

3、尽可能降低维护成本,保证对spark依赖的松耦合。

新的计算引擎

Hive的用户可以通过hive.execution.engine来设置计算引擎,目前该参数可选的值为mr何tez,为了实现Hive on spark,我们将spark作为参数第三个选项。要开启Hive on Spark模式,用户仅需将这个参数设置为Spark即可。

以Hive的表作为RDD

Spark以分布式可靠数据集合作为其数据抽象,因此我们需要将Hive的表转换为RDD以便spark处理。本质上,hive表和spark的HadoopRDD都是HDFS上的一组文件,通过inputFormat和RecordReader读取其中的数据,因此这个转化为自然而然的。

使用Hive原语

这里主要是指Hive的操作符对数据进行处理。Spark为RDD提供了一些列的转换Transformation,其中有些转换也是面向SQL的,比如groupByKey、join等如果使用这些转换(就如Shark所做的那样),那就意味着我们要重新实现一些Hive已有的功能;而且当Hive增加新的功能时,我们需要响应的Hive on spark模式。有鉴于此,我们选择将Hive的操作包装为Function,然后应用到RDD上,这样,我们只需要依赖较少的集中RDD的转换,而主要的计算逻辑任由Hive提供。

由于使用了Hive的原语,因此我们需要显示的调用一些Transformation来实现Shuffle的功能。下表中列举了Hive on Spark使用的所有转换。

Transformation                                        功能                                                         应用场景

mapPartitionsToPair                   将function应用到RDD的每个partition上           主要的计算逻辑

union                     返回两个RDD的联合                  多表查询

groupByKey        按照key对RDD进行group,group后的结果不保证有序,使用Hash Partitioner      Shuffle 使用用不要求排序的情况

sortByKey           按照key对RDD进行全局排序,使用Range Partitioner     Shuffle时使,适用于全局排序的查询,且由于使用了Range Partitioner,因此可以用多个reducer来实现全局排序

repartitionAndSortWithinPartitions   对RDD进行重新分区,并对每个分区进行排序,使用hash Partition  shuffle时使用,适用于需要排序的情况


repartitionAndSortWithinPartitions简单说明,这个功能由SPARK-2978引入,目的是提供一种MapReduce风格的Shuffle.虽然sortBySort也提供了排序的功能,但某些情况下我们并不需要全局有序,另外其使用的Range Partitioner对于某些Hive查询并不适用。

物理执行计划:

通过SparkCompiler将Operator Tree转换为Task Tree,其中需要提交给spark执行的任务即为sparkTask,不同于MapReduce中Map+Reduce的两阶段执行模式,spark采用DAG的执行模式,因此一个sparktask包含了一个表示RDD转换的DAG,我们将这个DAG包装为SparkWork。执行SparkTask时,就根据SparkWork所表示的DAG计算出最终的RDD,然后通过RDD的foreachAsync来出阿发原酸,使用foreachAsync是因为我们使用了Hive原语,因此不需要RDD返回结果;此外foreachAsync异步提交任务便于我们对任务进行监控。

SparkContext声明周期

sparkContext是用户与spark集群进行交互的接口,HIve on spark应该为每个用户会话创建一个SparkContext。但是Spark目前的使用方式假设SparkContext的生命周期是spark应用级别的,而且目前同一个JVM中不能创建多个SparkContext,这明显无法满足HiveServer2的应用场景,因为多个客户端需要通过一个HiveServer2来提供服务。鉴于此,我们需要在单独JVM中启动SparkContext,并通过RPC与远程的SparkContext进行通信。


任务监控与统计信息收集

spark提供了SparkListener接口来监听任务执行期间的各种事件,因此我们可以实现一个Listener来监控任务执行进度以及收集任务级别的统计的统计信息(目前任务级别的统计由sparkListener采集,任务进度则由spark提供专门的api来监控)另外Hive还提供了Operatior级别的统计数据信息,比如读取的行数等。在MapReduce模式下,这些信息通过Hadoop Counter收集。我们可以使用Spark提供的Accumulator来实现该功能。


测试

除了一般的单元测试以外,hive还提供了Qfile Test,即圆形一些事先定义好的查询,并根据结果判断测试是否通过。HIve on spark的Qfile Test应该尽可能接近真实的spark部署环境。目前我们采用的是local-cluster方式(该不是)。。。。

实现细节:

sparkTask的生成执行



我们通过一个例子看下一个简单的两表JOIN查询如何被转换成为sparkTask并被执行。下图左半部分展示了这个查询的operation tree,以及该operation Tree如何被转化成SparkTask;右半部分展示了该SparkTask执行时如何得到最终的RDD并通过foreachAsync提交spark任务

SparkCompiler遍历Operator Tree,将其划分为不同的MapWork和reducework,Mapwork为根节点,总是由TableScanOperator(hive中对表进行扫描操作符)开始;后续的work均为reducework。ReduceSinkOperator(Hive中进行Shuffle输出的操作符)用来标记两个Work之间的界限,出现ReduceSinkOperator表示当前work到下一个work之间的数据需要进行shuffle,因此当我们发现FileSinkOperator(hive中将结果输出到文件的操作符)的work为叶子节点。与MapReduce最大的不同在于,我们并不要求ReduceWork一定是叶子节点,即ReduceWork之后可以连接更多的reducework,并在同一个SParkTask中执行。

这个查询的OperatorTree 被转化为两个MapWork和一个ReduceWork。在执行SparkTask时,首先根据MapWork来生成最底层的HadoopRDD,然后将各个MapWork和ReduceWork包装成Function应用到RDD上,在有依赖的Work之间,需要显示的调用shuffle转换,具体选用哪种shuffle则根据查询的类型来确定。另外,由于这个例子涉及多表查询,因此在shuffle之前还要对RDD进行Union,经过一系列的转换后,得到最终的RDD,并通过foreachAsync提交到Spark集群上进行计算。



运行模式

RSC工作原理

Hive on Spark支持两种运行模式;本地和远程。当用户吧Spark Master URL设置为local时,采用本地模式;其余情况则采用远程模式。本地模式下,SparkContext与客户端运行在同一个JVM中,远程模式下,SparkContext运行在一个独立的JVM中,本地模式提供了主要为了调试,一般用户不应该选择该模式。因此我们这里主要介绍远程模式(Remote SparkContext。RSC)


用户的每个Session都会创建一个SparkClient,SparkClient会启动RemoteDriver进程,并由RemoteDriver创建SparkContext。SparkTask执行时,通过session提交任务,任务的主体就是对应的SparkWork。SparkClient将任务提交给RemoteDriver,并返回一个SparkJobRef,通过该SparkJobRef,客户端可以监控任务的进度,进行错误处理,以及采集统计信息等。由于最终的RDD计算没有返回结果,因此客户端只需要监控执行进度而不需要处理返回值。RemoteDriver通过SparkListenner收集任务级别的统计数据,通过Accumulator收集Operator级别的统计数据(Accumulator被包装为SparkCounter),并在任务结束时,返回给SparkClient。

SparkClient与RemoteDriver之间通过Netty的RPC进行通信。除了提交任务,SparkClient还提供了诸如添加jar包,获取集群信息的接口。如果客户端需要使用更一般的SparkContext的功能,可以自定义一个任务并通过SparkClient发送到Driver上执行。

理论上来说,HIve on Spark对Spark集群的部署方式没有特别的要求,除了local以外,RemoteDriver可以连接到任务的Spark集群来执行任务。在我们的测试中,HIve on Spark在Standalone和spark on yarn的计算上都能正常工作(需要动态添加jar包的查询在yarn-cluster模式下还不能运行)

优化

Map join

Map join是Hive中一个重要的优化,其原理是,如果参与join比较小的表可以存放如内存,因为这些小表在内存中生成Hash Table,这样较大的表只需要通过一个MapWork呗扫描一次,然后与内存中的Hash Table进行join了,生气了Shuffle和ReduceWork的开销。在MapReduce模式下,通过一个在客户端本地执行的任务来小表生成Hash table,并保存在文件系统上。后续的MapWork首先将Hash Table上传至Distibuted Cache中,最后只要读取大表和Distributed cache中的数据进行join就可以。

Hive on spark 对Map Join的实现与Map Reduce不同。当初我们考虑使用spark提供的广播功能来把小表的Hash table分发到各个节点上。使用广播的有点是spark采用高效的广播算法,其性能应该优于使用distributed cache。而使用广播的缺点会为Driver何计算及诶点带来很大的内存开销。为了使用广播,Hash table的数据需要先被传送到Driver端,然后由Driver进行广播;而且即使在广播之后,Driver仍需保留部分数据,以便应对计算节点的错误,虽然支持spill,但广播数据仍会加剧Driver的内存压力。此外广播相对的开发成本比较高,不利于对已有的代码的服用。

因此Hive on Spark选择类似于Distributed cache的方式来实现Map join,而且为小表生成Hash table的任务可以分布式的执行,进一步减轻客户端的压力。

不同于MapReduce,对于Hive on spark而言,LocalWork只是为了提供一些优化时必要的信息,并不会真正被执行。对于小表的扫描以独立的SparkTask分布式地执行,为此,我们也实现了能够分布式运行HashTableSinkOperator(Hive中输出小表Hash Table的操作符),其主要原理是通过提高HDFS Replication Factor的方式,是的生成的HashTable能够被每个节点在本地访问。

虽然目前采取了类似Distributed的这种实现方式,但如果再后期的测试中发现广播的方式确实能带来较大的性能提升,而且其引入的内存开销可以被接受,我们也会考虑改用广播来实现Map Join

Table cache

Spark的一个优势是可以利用充分利用内存,允许用户显示地把一个RDD保存到内存或者磁盘上,以便于在多次访问时候提高性能。另外在目前的RDD转换模式弘,一个RDD的数据是无法同时被多个下游使用,当一个RDD需要通过不同的转换得到不同的子节点时,就要被计算多次。这时我们应该使用cache来避免重复计算。

在Shark和SparkSQL中,都允许用户显式吧一张表cache来提高对该表的查询性能。比如以下查询,在





在这种情况下,对应的SparkWork中,一个MapWork/ReduceWork会有多个下游的Work,如果不进行cache,那么共享的数据就会被计算多次。为了避免这种情况,我们会将这些MapWork/ReduceWork复制成多个,每个对应一个下游的Work,并对其共享的数据进行cache(由于IOContext的同步问题,该工功能尚未完成)

更为一般的应用场景是一张表在查询中被使用了多次的情况,Hive on spark 目前还不会针对这种查询进行cache,


初步性能测试:

测试集群由10台虚拟机组成,测试数据为320GB的TPC-DS数据。目前的测试用例为6条,包含了自定义的查询以及TPC-DS数据集合。目前的测试用例为6条,包含了自定义的查询以及TPC-DS中两条查询。由于Hive主要用于处理ETL查询,因此我们在TPC-DS中选取用例时,选用较为接近ETL查询用例(TPC-DS的用例主要针对交互式查询,impala,sparkSQL等引擎更合适此类查询),主要针对Hive on Spark 和Hive on Tez进行性能对比


Hive on Spark vs. Hive on Tez

图中横坐标为各个测试用例,纵坐标为所用时间,以秒为单位。

总结:

Hive on spark由多家公司协作开发,从项目开始以来,受到社区的广泛关注。

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 212,222评论 6 493
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 90,455评论 3 385
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 157,720评论 0 348
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 56,568评论 1 284
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 65,696评论 6 386
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 49,879评论 1 290
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 39,028评论 3 409
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 37,773评论 0 268
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 44,220评论 1 303
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,550评论 2 327
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,697评论 1 341
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 34,360评论 4 332
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 40,002评论 3 315
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,782评论 0 21
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,010评论 1 266
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 46,433评论 2 360
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 43,587评论 2 350

推荐阅读更多精彩内容