Spark框架深度理解二：生态圈

前言

由于Spark框架大多都搭建在Hadoop系统之上&＃xff0c;要明白Spark核心运行原理还是得对Hadoop体系有个熟悉的认知。从Hadoop1.0到Hadoop2.0架构的优化和发展探索详解这篇博客大家可以先去温习一下Hadoop整个体系&＃xff0c;然后再来了解Spark框架会更有效率。

本来想直接写一篇缘由优缺点以及生态圈和运行架构与原理的&＃xff0c;发现篇幅实在是太长了&＃xff0c;索性分两篇&＃xff1a;

上篇&＃xff1a;Spark框架深度理解一&＃xff1a;开发缘由及优缺点

Spark生态圈

Spark生态圈以Spark Core为核心&＃xff0c;从HDFS、Amazon S3和HBase等持久层读取数据&＃xff0c;以MESS、YARN和自身携带的Standalone为资源管理器调度Job完成Spark应用程序的计算。 这些应用程序可以来自于不同的组件&＃xff0c;如Spark Shell/Spark Submit的批处理、Spark Streaming的实时处理应用、Spark SQL的即席查询、BlinkDB的权衡查询、MLlib/MLbase的机器学习、GraphX的图处理和SparkR的数学计算等等。

 1. Spark Core

Saprk核心&＃xff0c;提供底层框架及核心支持。

• 提供了有向无环图&＃xff08;DAG&＃xff09;的分布式并行计算框架&＃xff0c;并提供Cache机制来支持多次迭代计算或者数据共享&＃xff0c;大大减少迭代计算之间读取数据局的开销&＃xff0c;这对于需要进行多次迭代的数据挖掘和分析性能有很大提升
• 在Spark中引入了RDD (Resilient Distributed Dataset) 的抽象&＃xff0c;它是分布在一组节点中的只读对象集合&＃xff0c;这些集合是弹性的&＃xff0c;如果数据集一部分丢失&＃xff0c;则可以根据“血统”对它们进行重建&＃xff0c;保证了数据的高容错性&＃xff1b;
• 移动计算而非移动数据&＃xff0c;RDD Partition可以就近读取分布式文件系统中的数据块到各个节点内存中进行计算
• 使用多线程池模型来减少task启动开稍
• 采用容错的、高可伸缩性的akka作为通讯框架

2.SparkStreaming

SparkStreaming是一个对实时数据流进行高通量、容错处理的流式处理系统&＃xff0c;可以对多种数据源&＃xff08;如Kdfka、Flume、Twitter、Zero和TCP 套接字&＃xff09;进行类似Map、Reduce和Join等复杂操作&＃xff0c;并将结果保存到外部文件系统、数据库或应用到实时仪表盘。

Spark Streaming构架

计算流程&＃xff1a;Spark Streaming是将流式计算分解成一系列短小的批处理作业。这里的批处理引擎是Spark Core&＃xff0c;也就是把Spark Streaming的输入数据按照batch size&＃xff08;如1秒&＃xff09;分成一段一段的数据&＃xff08;Discretized Stream&＃xff09;&＃xff0c;每一段数据都转换成Spark中的RDD&＃xff08;Resilient Distributed Dataset&＃xff09;&＃xff0c;然后将Spark Streaming中对DStream的Transformation操作变为针对Spark中对RDD的Transformation操作&＃xff0c;将RDD经过操作变成中间结果保存在内存中。整个流式计算根据业务的需求可以对中间的结果进行叠加或者存储到外部设备。下图显示了Spark Streaming的整个流程。

容错性&＃xff1a;对于流式计算来说&＃xff0c;容错性至关重要。首先我们要明确一下Spark中RDD的容错机制。每一个RDD都是一个不可变的分布式可重算的数据集&＃xff0c;其记录着确定性的操作继承关系&＃xff08;lineage&＃xff09;&＃xff0c;所以只要输入数据是可容错的&＃xff0c;那么任意一个RDD的分区&＃xff08;Partition&＃xff09;出错或不可用&＃xff0c;都是可以利用原始输入数据通过转换操作而重新算出的。  

对于Spark Streaming来说&＃xff0c;其RDD的传承关系如下图所示&＃xff0c;图中的每一个椭圆形表示一个RDD&＃xff0c;椭圆形中的每个圆形代表一个RDD中的一个Partition&＃xff0c;图中的每一列的多个RDD表示一个DStream&＃xff08;图中有三个DStream&＃xff09;&＃xff0c;而每一行最后一个RDD则表示每一个Batch Size所产生的中间结果RDD。我们可以看到图中的每一个RDD都是通过lineage相连接的&＃xff0c;由于Spark Streaming输入数据可以来自于磁盘&＃xff0c;例如HDFS&＃xff08;多份拷贝&＃xff09;或是来自于网络的数据流&＃xff08;Spark Streaming会将网络输入数据的每一个数据流拷贝两份到其他的机器&＃xff09;都能保证容错性&＃xff0c;所以RDD中任意的Partition出错&＃xff0c;都可以并行地在其他机器上将缺失的Partition计算出来。这个容错恢复方式比连续计算模型&＃xff08;如Storm&＃xff09;的效率更高。

实时性&＃xff1a;对于实时性的讨论&＃xff0c;会牵涉到流式处理框架的应用场景。Spark Streaming将流式计算分解成多个Spark Job&＃xff0c;对于每一段数据的处理都会经过Spark DAG图分解以及Spark的任务集的调度过程。对于目前版本的Spark Streaming而言&＃xff0c;其最小的Batch Size的选取在0.5~2秒钟之间&＃xff08;Storm目前最小的延迟是100ms左右&＃xff09;&＃xff0c;所以Spark Streaming能够满足除对实时性要求非常高&＃xff08;如高频实时交易&＃xff09;之外的所有流式准实时计算场景。

扩展性与吞吐量&＃xff1a;Spark目前在EC2上已能够线性扩展到100个节点&＃xff08;每个节点4Core&＃xff09;&＃xff0c;可以以数秒的延迟处理6GB/s的数据量&＃xff08;60M records/s&＃xff09;&＃xff0c;其吞吐量也比流行的Storm高2&＃xff5e;5倍&＃xff0c;图是Berkeley利用WordCount和Grep两个用例所做的测试&＃xff0c;在Grep这个测试中&＃xff0c;Spark Streaming中的每个节点的吞吐量是670k records/s&＃xff0c;而Storm是115k records/s。

3.Spark SQL

Shark是SparkSQL的前身&＃xff0c;它发布于3年前&＃xff0c;那个时候Hive可以说是SQL on Hadoop的唯一选择&＃xff0c;负责将SQL编译成可扩展的MapReduce作业&＃xff0c;鉴于Hive的性能以及与Spark的兼容&＃xff0c;Shark项目由此而生。

Shark即Hive on Spark&＃xff0c;本质上是通过Hive的HQL解析&＃xff0c;把HQL翻译成Spark上的RDD操作&＃xff0c;然后通过Hive的metadata获取数据库里的表信息&＃xff0c;实际HDFS上的数据和文件&＃xff0c;会由Shark获取并放到Spark上运算。Shark的最大特性就是快和与Hive的完全兼容&＃xff0c;且可以在shell模式下使用rdd2sql()这样的API&＃xff0c;把HQL得到的结果集&＃xff0c;继续在scala环境下运算&＃xff0c;支持自己编写简单的机器学习或简单分析处理函数&＃xff0c;对HQL结果进一步分析计算。

在2014年7月1日的Spark Summit上&＃xff0c;Databricks宣布终止对Shark的开发&＃xff0c;将重点放到Spark SQL上。Databricks表示&＃xff0c;Spark SQL将涵盖Shark的所有特性&＃xff0c;用户可以从Shark 0.9进行无缝的升级。在会议上&＃xff0c;Databricks表示&＃xff0c;Shark更多是对Hive的改造&＃xff0c;替换了Hive的物理执行引擎&＃xff0c;因此会有一个很快的速度。然而&＃xff0c;不容忽视的是&＃xff0c;Shark继承了大量的Hive代码&＃xff0c;因此给优化和维护带来了大量的麻烦。随着性能优化和先进分析整合的进一步加深&＃xff0c;基于MapReduce设计的部分无疑成为了整个项目的瓶颈。因此&＃xff0c;为了更好的发展&＃xff0c;给用户提供一个更好的体验&＃xff0c;Databricks宣布终止Shark项目&＃xff0c;从而将更多的精力放到Spark SQL上。

Spark SQL允许开发人员直接处理RDD&＃xff0c;同时也可查询例如在 Apache Hive上存在的外部数据。Spark SQL的一个重要特点是其能够统一处理关系表和RDD&＃xff0c;使得开发人员可以轻松地使用SQL命令进行外部查询&＃xff0c;同时进行更复杂的数据分析。除了Spark SQL外&＃xff0c;Michael还谈到Catalyst优化框架&＃xff0c;它允许Spark SQL自动修改查询方案&＃xff0c;使SQL更有效地执行。

还有Shark的作者是来自中国的博士生辛湜&＃xff08;Reynold Xin&＃xff09;&＃xff0c;也是Spark的核心成员&＃xff0c;具体信息可以看他的专访Spark核心开发者&＃xff1a;性能超Hadoop百倍&＃xff0c;算法实现仅有其1/10或1/100。

Spark SQL的特点:

• 引入了新的RDD类型SchemaRDD&＃xff0c;可以象传统数据库定义表一样来定义SchemaRDD&＃xff0c;SchemaRDD由定义了列数据类型的行对象构成。SchemaRDD可以从RDD转换过来&＃xff0c;也可以从Parquet文件读入&＃xff0c;也可以使用HiveQL从Hive中获取。
• 内嵌了Catalyst查询优化框架&＃xff0c;在把SQL解析成逻辑执行计划之后&＃xff0c;利用Catalyst包里的一些类和接口&＃xff0c;执行了一些简单的执行计划优化&＃xff0c;最后变成RDD的计算
• 在应用程序中可以混合使用不同来源的数据&＃xff0c;如可以将来自HiveQL的数据和来自SQL的数据进行Join操作。

Shark的出现使得SQL-on-Hadoop的性能比Hive有了10-100倍的提高&＃xff0c; 虽然没有Shark相对于Hive那样瞩目地性能提升&＃xff0c;但也表现得非常优异&＃xff0c;如下图所示&＃xff1a;

为什么sparkSQL的性能会得到怎么大的提升呢&＃xff1f;主要sparkSQL在下面几点做了优化&＃xff1a;

• 内存列存储&＃xff08;In-Memory Columnar Storage&＃xff09; sparkSQL的表数据在内存中存储不是采用原生态的JVM对象存储方式&＃xff0c;而是采用内存列存储&＃xff1b;
• 字节码生成技术&＃xff08;Bytecode Generation&＃xff09; Spark1.1.0在Catalyst模块的expressions增加了codegen模块&＃xff0c;使用动态字节码生成技术&＃xff0c;对匹配的表达式采用特定的代码动态编译。另外对SQL表达式都作了CG优化&＃xff0c; CG优化的实现主要还是依靠Scala2.10的运行时放射机制&＃xff08;runtime reflection&＃xff09;&＃xff1b;
• Scala代码优化 SparkSQL在使用Scala编写代码的时候&＃xff0c;尽量避免低效的、容易GC的代码&＃xff1b;尽管增加了编写代码的难度&＃xff0c;但对于用户来说接口统一。

4.BlinkDB

BlinkDB 是一个用于在海量数据上运行交互式 SQL 查询的大规模并行查询引擎&＃xff0c;它允许用户通过权衡数据精度来提升查询响应时间&＃xff0c;其数据的精度被控制在允许的误差范围内。为了达到这个目标&＃xff0c;BlinkDB 使用两个核心思想:

• 一个自适应优化框架&＃xff0c;从原始数据随着时间的推移建立并维护一组多维样本&＃xff1b;
• 一个动态样本选择策略&＃xff0c;选择一个适当大小的示例基于查询的准确性和&＃xff08;或&＃xff09;响应时间需求。

和传统关系型数据库不同&＃xff0c;BlinkDB是一个很有意思的交互式查询系统&＃xff0c;就像一个跷跷板&＃xff0c;用户需要在查询精度和查询时间上做一权衡&＃xff1b;如果用户想更快地获取查询结果&＃xff0c;那么将牺牲查询结果的精度&＃xff1b;同样的&＃xff0c;用户如果想获取更高精度的查询结果&＃xff0c;就需要牺牲查询响应时间。用户可以在查询的时候定义一个失误边界。

5.MLBase/MLlib

 MLBase是Spark生态圈的一部分专注于机器学习&＃xff0c;让机器学习的门槛更低&＃xff0c;让一些可能并不了解机器学习的用户也能方便地使用MLbase。MLBase分为四部分&＃xff1a;MLlib、MLI、ML Optimizer和MLRuntime。

• ML Optimizer会选择它认为最适合的已经在内部实现好了的机器学习算法和相关参数&＃xff0c;来处理用户输入的数据&＃xff0c;并返回模型或别的帮助分析的结果&＃xff1b;
• MLI 是一个进行特征抽取和高级ML编程抽象的算法实现的API或平台&＃xff1b;
• MLlib是Spark实现一些常见的机器学习算法和实用程序&＃xff0c;包括分类、回归、聚类、协同过滤、降维以及底层优化&＃xff0c;该算法可以进行可扩充&＃xff1b; MLRuntime 基于Spark计算框架&＃xff0c;将Spark的分布式计算应用到机器学习领域。

总的来说&＃xff0c;MLBase的核心是他的优化器&＃xff0c;把声明式的Task转化成复杂的学习计划&＃xff0c;产出最优的模型和计算结果。与其他机器学习Weka和Mahout不同的是&＃xff1a;

• MLBase是分布式的&＃xff0c;Weka是一个单机的系统&＃xff1b;
• MLBase是自动化的&＃xff0c;Weka和Mahout都需要使用者具备机器学习技能&＃xff0c;来选择自己想要的算法和参数来做处理&＃xff1b;
• MLBase提供了不同抽象程度的接口&＃xff0c;让算法可以扩充
• MLBase基于Spark这个平台

 6.GraphX

GraphX是Spark中用于图(e.g., Web-Graphs and Social Networks)和图并行计算(e.g., PageRank and Collaborative Filtering)的API,可以认为是GraphLab(C&＃43;&＃43;)和Pregel(C&＃43;&＃43;)在Spark(Scala)上的重写及优化&＃xff0c;跟其他分布式图计算框架相比&＃xff0c;GraphX最大的贡献是&＃xff0c;在Spark之上提供一栈式数据解决方案&＃xff0c;可以方便且高效地完成图计算的一整套流水作业。GraphX最先是伯克利AMPLAB的一个分布式图计算框架项目&＃xff0c;后来整合到Spark中成为一个核心组件。

GraphX的核心抽象是Resilient Distributed Property Graph&＃xff0c;一种点和边都带属性的有向多重图。它扩展了Spark RDD的抽象&＃xff0c;有Table和Graph两种视图&＃xff0c;而只需要一份物理存储。两种视图都有自己独有的操作符&＃xff0c;从而获得了灵活操作和执行效率。如同Spark&＃xff0c;GraphX的代码非常简洁。GraphX的核心代码只有3千多行&＃xff0c;而在此之上实现的Pregel模型&＃xff0c;只要短短的20多行。GraphX的代码结构整体下图所示&＃xff0c;其中大部分的实现&＃xff0c;都是围绕Partition的优化进行的。这在某种程度上说明了点分割的存储和相应的计算优化的确是图计算框架的重点和难点。

GraphX的底层设计有以下几个关键点。

1.对Graph视图的所有操作&＃xff0c;最终都会转换成其关联的Table视图的RDD操作来完成。这样对一个图的计算&＃xff0c;最终在逻辑上&＃xff0c;等价于一系列RDD的转换过程。因此&＃xff0c;Graph最终具备了RDD的3个关键特性&＃xff1a;Immutable、Distributed和Fault-Tolerant。其中最关键的是Immutable&＃xff08;不变性&＃xff09;。逻辑上&＃xff0c;所有图的转换和操作都产生了一个新图&＃xff1b;物理上&＃xff0c;GraphX会有一定程度的不变顶点和边的复用优化&＃xff0c;对用户透明。

2.两种视图底层共用的物理数据&＃xff0c;由RDD[Vertex-Partition]和RDD[EdgePartition]这两个RDD组成。点和边实际都不是以表Collection[tuple]的形式存储的&＃xff0c;而是由VertexPartition/EdgePartition在内部存储一个带索引结构的分片数据块&＃xff0c;以加速不同视图下的遍历速度。不变的索引结构在RDD转换过程中是共用的&＃xff0c;降低了计算和存储开销。

3.图的分布式存储采用点分割模式&＃xff0c;而且使用partitionBy方法&＃xff0c;由用户指定不同的划分策略&＃xff08;PartitionStrategy&＃xff09;。划分策略会将边分配到各个EdgePartition&＃xff0c;顶点Master分配到各个VertexPartition&＃xff0c;EdgePartition也会缓存本地边关联点的Ghost副本。划分策略的不同会影响到所需要缓存的Ghost副本数量&＃xff0c;以及每个EdgePartition分配的边的均衡程度&＃xff0c;需要根据图的结构特征选取最佳策略。目前有EdgePartition2d、EdgePartition1d、RandomVertexCut和CanonicalRandomVertexCut这四种策略。在淘宝大部分场景下&＃xff0c;EdgePartition2d效果最好。

7.SparkR

SparkR是AMPLab发布的一个R开发包&＃xff0c;使得R摆脱单机运行的命运&＃xff0c;可以作为Spark的job运行在集群上&＃xff0c;极大得扩展了R的数据处理能力。

parkR的几个特性&＃xff1a;

• 提供了Spark中弹性分布式数据集&＃xff08;RDD&＃xff09;的API&＃xff0c;用户可以在集群上通过R shell交互性的运行Spark job。
• 支持序化闭包功能&＃xff0c;可以将用户定义函数中所引用到的变量自动序化发送到集群中其他的机器上。
• SparkR还可以很容易地调用R开发包&＃xff0c;只需要在集群上执行操作前用includePackage读取R开发包就可以了&＃xff0c;当然集群上要安装R开发包。

8.Tachyon

Tachyon是一个高容错的分布式文件系统&＃xff0c;允许文件以内存的速度在集群框架中进行可靠的共享&＃xff0c;就像Spark和 MapReduce那样。通过利用信息继承&＃xff0c;内存侵入&＃xff0c;Tachyon获得了高性能。Tachyon工作集文件缓存在内存中&＃xff0c;并且让不同的 Jobs/Queries以及框架都能内存的速度来访问缓存文件”。因此&＃xff0c;Tachyon可以减少那些需要经常使用的数据集通过访问磁盘来获得的次数。Tachyon兼容Hadoop&＃xff0c;现有的Spark和MR程序不需要任何修改而运行。 

参阅&＃xff1a;

计算机内存和磁盘的关系

spark的前世今生

Spark及其生态圈简介