Shark简介

简介

Shark是一个新的数据分析系统，在集群上进行查询处理和复杂分析。Shark使用一种新的分布式内存抽象，为SQL查询和复杂分析函数提供了统一的运行引擎，并能够有效的容错。这种新的分布式内存抽象使得shark执行SQL查询的速度比Hive快100倍，执行机器学习算法比Hadoop快100倍。与之前的系统不同，shark在取得如此良好性能的同时尽可能的保留了类似于mapreduce的执行框架，以及细粒度的容错特性。为了有效的执行SQL查询，shark提供了面向列存储，查询中重计划等机制。Shark在速度上能够与MPP分析数据库相当，同时又具有MPP分析数据库不具备的容错和复杂分析的能力。

Shark基本上就是在Spark的框架基础上提供和Hive一样的HiveQL命令接口，为了最大程度的保持和Hive的兼容性，Shark使用了Hive的API来实现query Parsing和Logic Plan generation，最后的PhysicalPlan execution阶段用Spark代替Hadoop MapReduce。通过配置Shark参数，Shark可以自动在内存中缓存特定的RDD，实现数据重用，进而加快特定数据集的检索。同时，Shark通过UDF用户自定义函数实现特定的数据分析学习算法，使得SQL数据查询和运算分析能结合在一起，最大化RDD的重复使用。

系统概述

Shark既支持SQL查询处理，也支持机器学习函数。Shark是与Hive兼容的，用户可以无需修改Hive的查询以及数据就能在shark上执行。

由于与Hive的兼容，shark可以在任何支持Hadoop存储API的系统上进行查询，包括HDFS和Amazon S3等。Shark也支持多种数据格式，例如：text，json，xml等。Shark继承了Hive的schema-on-read和嵌套数据类型。

此外，用户可以选择将部分数据存储到shark的内存中，以便进行快速分析处理。

上图显示了shark集群的体系结构，包括一个master节点和一组worker节点。数据仓库的元数据存储在一个外部事务性数据库中。Shark是建立在Spark之上的，Spark是一个类似于Mapreduce的集群计算引擎。当一个查询提交到master时，shark将该查询转换成一个用RDD表示的操作树。这些RDD被Spark转化成一张任务的图，worker节点来执行这些任务。

上图显示了shark的体系结构，特点如下：

1、物理计划在Spark上执行；

2、依靠Spark的快速执行、容错性及RDD；

3、尽可能的充分利用Hive的代码：将Hive生成的逻辑计划转换为Spark的执行图（execution graph）；

4、与Hive的兼容性：无需修改就可以利用hive的metadata在HDFS上执行HiveQL。

在RDD上执行SQL查询

Shark与传统的RDBMS一样，执行SQL查询一般都经过以下3个过程：解析查询语句、生成逻辑计划、生成物理计划。给定一个查询，Shark使用Hive的查询编译器解析查询语句并生成对应的抽象语法树，然后将这棵树转化为逻辑计划及基本的逻辑优化（比如，预测下推），以上这些，Shark和Hive都使用了相同的方法。Hive将把操作转化为有一系列MapReduce程序组成的物理计划，而对于Shark，它的优化器将应用额外的优化规则（如，对于关键字LIMIT对应的优化规则）到各个独立的分区（individual partitions）中，并且创建由一序列在RDDs上的转化操作构成的物理计划，而不是MapReduce Jobs，然后利用在Spark中已经实现的operators（如，map、reduce）或在Shark中实现的operators（如 broadcast joins）执行相应的物理计划。

系统拓展

尽管通过这些基本的方法可以在Spark上执行SQL语句，但怎样做到高效率是一个挑战。UDF（用户自定义函数）的流行以及复杂的分析函数在Shark上的使用，很难再编译的时候就确定相应的优化查询计划，特别在没有经过ETL处理的新数据上。另外，尽管有了这样一个计划，愚蠢的在Spark或其他MapReduce环境下也是低效的。为了高效的存储相关的数据以及执行SQL，对Spark引擎进行了一些如下拓展。

Ø Partial DAG Execution（PDE）

为了在分布式数据集上支持动态查询优化，拓展了Spark支持PDE，这项技术允许使用在运行时收集的统计数据，动态的改变查询计划。目前，PDE主要用在Shuffle操作阶段边界上，这里将进行数据的交换及重新分区，都是Shark上典型的高消耗操作。默认情况下，在shuffle之前，Spark将在内存中实例化map任务的输出，如果有必要才输出到磁盘上，reduce任务将随后获取这些输出。

PDE通过以下两步修改这种机制。首先，在实例化map输出的时收集总体可定制的统计信息和每个分区的粒度；然后，将依赖这些统计值，通过选择不同的operators或改变它们的参数（如，并行度）来修改DAG。

Ø Columnar Memory Store

Shark在Spark的内存存储机制上实现了基于列的内存存储。Shark使用JVM原始的数组存储原始类型（primitive types）的所有列，Hive支持的复杂数据类型（如，map、array），将被序列化并由关联着的字节数组进行存储。每一列都只创建了一个JVM对象，能够实现高效的空间利用、GCs和压缩数据，减轻了deserialize的负担。

Ø Distributed Data Loading

除了查询的执行，Shark还使用Spark的执行引擎实现了分布式数据加载。在加载的过程中，表将被分割成小的partitions，这些partitions将被不同的tasks根据加载模式，提取相应的数据并以列的形式存储在内存中。每一个数据加载的task将跟踪metadata来决定分区中的每一列是否应该压缩，如：若某列不同的值低于某个阈值，对应的数据加载task将在此列上进行压缩处理。每个task根据加载数据的情况，选择合适的处理优化策略，而不是所有的task都统一使用共同的策略，保证在加载过程中实现最大的并行度。需要注意的是，RDD的血统并不会记录这些处理的策略。

Ø Data Co-partitioning

Shark为了在后面的查询中快速的实现join操作，在数据加载过程中，允许在共同的主键上进行co-partitioning。在分布式文件系统（如，HDFS）中，存储系统基于模式判别，不能实现co-partitioning。在join两个co-partitioned tables时，Shark的优化器将会创建一个避免了高耗shuffle的DAG并且代替map任务执行join。在Shark中可以使用DISTRIBUTE BY进行实现：

CREATE TABLE l_mem TBLPROPERTIES ("shark.cache"=true)

AS SELECT * FROM lineitem DISTRIBUTE BY L_ORDERKEY;

CREATE TABLE o_mem TBLPROPERTIES ("shark.cache"=true,"copartition"="l_mem") AS SELECT * FROM order DISTRIBUTE BY O_ORDERKEY;

Ø Partition Statistics and Map Pruning

存储文件系统中的数据，很有可能在一列或几列上存在在聚集的性质，Map Pruning就是建立在具有聚集特性的数据上进行数据分区修剪的过程。Shark的内存将数据存储成小的分区（block），每个block都包含了一个或几个逻辑上具有聚集性质的组，而Shark可以避免扫描那些不在查询范围内的block。

为了充分利用这些在某些列上具有聚集特性的数据，Shark的内存存储了每个worker在数据加载过程中记录的统计信息，包括每一列值的范围。如果某列上值不同的数较少，还将统计这些不同的值，这些统计值将在master中汇总，实现查询过程中修剪分区。

支持机器学习

Shark的一个重要设计目标就是能处理高效SQL查询和复杂的机器学习。Shark遵循将计算移到数据处的原则，支持机器学习。这样设计的原因是选择了Spark作为执行引擎、RDD作为operators操作的主要数据结构。Shark除了执行SQL查询并返回其结果外，还允许查询返回代表了查询计划的RDD，用户可以使用返回的RDD进行分布式计算。

Shark会在集群上自动的并行执行上述的map、mapRows和reduce函数，并且master程序将收集reduce函数的输出进行更新操作。目前Shark提供一些基本的机器学习算法，包括linear regression、logistic regression和k-means。在大部分情况下，用户只需要实现一个mapRows函数提取待分析的数据，然后调用上述提供的算法。