1 Spark简介

1.1 引言

行业正在广泛使用Hadoop来分析他们的数据集。原因是Hadoop框架基于简单的编程模型(MapReduce)，它使计算解决方案具有可扩展性，灵活性，容错性和成本效益。在这里，主要关注的是在查询之间的等待时间和运行程序的等待时间方面保持处理大型数据集的速度。

Spark由Apache Software Foundation引入，用于加速Hadoop计算计算软件过程。

与普遍看法相反，Spark不是Hadoop的修改版本，并且实际上并不依赖于Hadoop，因为它有自己的集群管理。Hadoop只是实现Spark的方法之一。

Spark以两种方式使用Hadoop - 一种是存储，另一种是处理。由于Spark有自己的集群管理计算，因此它仅将Hadoop用于存储目的。

1.2 Apache Spark

Apache Spark是一种闪电般快速的集群计算技术，专为快速计算而设计。它基于Hadoop MapReduce，它扩展了MapReduce模型，以便有效地将其用于更多类型的计算，包括交互式查询和流处理。Spark的主要特性是其内存中的集群计算，可以提高应用程序的处理速度。

Spark旨在涵盖广泛的工作负载，如批处理应用程序，迭代算法，交互式查询和流式处理。除了在相应的系统中支持所有这些工作负载之外，它还减少了维护单独工具的管理负担。

1.3 Apache Spark的演变

Spark是2009年在加州大学伯克利分校的Matei Zaharia的AMPLab开发的Hadoop子项目之一。它是在2010年根据BSD许可证开源的。它于2013年被捐赠给Apache软件基金会，现在Apache Spark已成为2014年2月的顶级Apache项目。

1.4 Apache Spark的功能

Apache Spark具有以下功能。

速度 - Spark有助于在Hadoop集群中运行应用程序，内存速度提高100倍，在磁盘上运行速度提高10倍。这可以通过减少对磁盘的读/写操作次数来实现。它将中间处理数据存储在存储器中。

支持多种语言 --Spark提供Java，Scala或Python内置API。因此，您可以用不同的语言编写应用程序。Spark提供了80个用于交互式查询的高级操作员。

高级分析 - Spark不仅支持'Map'和'reduce'。它还支持SQL查询，流数据，机器学习(ML)和图形算法。

1.5 Spark建立在Hadoop上

下图显示了如何使用Hadoop组件构建Spark的三种方法。

Spark部署有三种方式，如下所述。

独立 - Spark Standalone部署意味着Spark占据了HDFS(Hadoop分布式文件系统)之上的位置，并且明确地为HDFS分配了空间。在这里，Spark和MapReduce将并排运行，以涵盖集群上的所有火花作业。

Hadoop Yarn - Hadoop Yarn部署意味着，简单地说，在Yarn上运行spark，无需任何预安装或root访问。它有助于将Spark集成到Hadoop生态系统或Hadoop堆栈中。它允许其他组件在堆栈上运行。

MapReduce(SIMR)中的 Spark - 除了独立部署之外，MapReduce中的Spark还用于启动spark作业。使用SIMR，用户可以启动Spark并使用其shell而无需任何管理访问权限。

1.6 Spark的组件

下图描绘了Spark的不同组件

Apache Spark Core

Spark Core是Spark平台的基础通用执行引擎，所有其他功能都基于此构建。它在外部存储系统中提供内存计算和引用数据集。

Spark SQL

Spark SQL是Spark Core之上的一个组件，它引入了一个名为SchemaRDD的新数据抽象，它为结构化和半结构化数据提供支持。

Spark Streaming

Spark Streaming利用Spark Core的快速调度功能来执行流分析。它以小批量方式提取数据，并对这些小批量数据执行RDD(弹性分布式数据集)转换。

MLlib(机器学习库)

由于基于分布式内存的Spark架构，MLlib是Spark之上的分布式机器学习框架。根据基准测试，MLlib开发人员针对交替最小二乘(ALS)实现进行了测试。Spark MLlib的速度是基于Hadoop磁盘的Apache Mahout版本的9倍(在Mahout获得Spark界面之前)。

GraphX

GraphX是Spark上的分布式图形处理框架。它提供了一个表达图形计算的API，可以使用Pregel抽象API对用户定义的图形进行建模。它还为此抽象提供了优化的运行时。

2 Apache Spark - RDD

2. 1 Resilient Distributed Datasets

弹性分布式数据集(RDD )是Spark的基本数据结构。它是一个不可变的分布式对象集合。RDD中的每个数据集被划分为逻辑分区，其可以在集群的不同节点上计算。RDD可以包含任何类型的Python，Java或Scala对象，包括用户定义的类。

形式上，RDD是只读的分区记录集合。可以通过对稳定存储或其他RDD上的数据的确定性操作来创建RDD。RDD是一个容错的容错集合，可以并行操作。

有两种方法可以创建RDD - 在驱动程序中并行化现有集合，或在外部存储系统中引用数据集，例如共享文件系统，HDFS，HBase或提供Hadoop输入格式的任何数据源。

Spark利用RDD的概念来实现更快，更高效的MapReduce操作。让我们首先讨论MapReduce操作是如何发生的以及为什么它们不那么有效。

2. 2 MapReduce中的数据共享速度很慢

MapReduce被广泛用于在集群上使用并行分布式算法处理和生成大型数据集。它允许用户使用一组高级操作符编写并行计算，而不必担心工作分配和容错。

不幸的是，在大多数当前框架中，在计算之间重用数据的唯一方法(Ex-两个MapReduce作业之间)是将其写入外部稳定存储系统(Ex-HDFS)。虽然这个框架提供了许多用于访问集群计算资源的抽象，但用户仍然需要更多。

这两个迭代和交互式应用程序需要跨并行作业更快速的数据共享。由于复制，序列化和磁盘IO，MapReduce中的数据共享速度很慢。关于存储系统，大多数Hadoop应用程序，他们花费超过90％的时间进行HDFS读写操作。

2. 3 MapReduce上的迭代操作

在多阶段应用程序中跨多个计算重用中间结果。下图说明了在MapReduce上执行迭代操作时当前框架的工作原理。由于数据复制，磁盘I / O和序列化，这会导致大量开销，这会使系统变慢。

2. 4 MapReduce上的交互操作

用户对同一数据子集运行即席查询。每个查询都将在稳定存储上执行磁盘I / O，这可以控制应用程序执行时间。

下图说明了在MapReduce上执行交互式查询时当前框架的工作原理。

MapReduce上的交互操作

2. 5 使用Spark RDD进行数据共享

由于复制，序列化和磁盘IO，MapReduce中的数据共享速度很慢。大多数Hadoop应用程序，他们花费超过90％的时间进行HDFS读写操作。

认识到这个问题，研究人员开发了一个名为Apache Spark的专用框架。火花的关键思想是- [R esilient d istributed d atasets(RDD); 它支持内存处理计算。这意味着，它将内存状态存储为作业中的对象，并且对象可在这些作业之间共享。内存中的数据共享比网络和磁盘快10到100倍。

现在让我们试着找出Spark RDD中迭代和交互操作的发生方式。

2. 6 Spark RDD的迭代操作

下面给出的插图显示了Spark RDD的迭代操作。它将中间结果存储在分布式存储器中而不是稳定存储(磁盘)中，从而使系统更快。

注 - 如果分布式内存(RAM)不足以存储中间结果(JOB的状态)，则它会将这些结果存储在磁盘上。

Spark RDD的迭代操作

2. 7 Spark RDD的交互操作

此图显示了Spark RDD上的交互操作。如果对同一组数据重复运行不同的查询，则可以将此特定数据保存在内存中以获得更好的执行时间。

Spark RDD的交互操作

默认情况下，每次对其执行操作时，都可以重新计算每个转换后的RDD。但是，您也可以在内存中保留 RDD，在这种情况下，Spark会在群集上保留元素，以便在下次查询时更快地访问。还支持在磁盘上保留RDD或在多个节点上复制。

3 Spark安装

这个步骤不再赘述，google一下相关的文章比较多，我推荐一篇文章Spark学习之路 (五)Spark伪分布式安装。按照上面的步骤一步一步来，不要跳过步骤，然后安装时比如路径或者ip端口可以根据自己的实际情况更改。

安装成功之后，执行sudo sbin/start-all.sh,可以看到如下进程：

通过http://192.168.31.135:8080，查看Web界面

4 Spark 核心编程

Spark Core是整个项目的基础。它提供分布式任务调度，调度和基本I / O功能。Spark使用称为RDD(弹性分布式数据集)的专用基础数据结构，它是跨机器分区的数据的逻辑集合。RDD可以通过两种方式创建; 一种是通过引用外部存储系统中的数据集，另一种是通过在现有RDD上应用转换(例如map，filter，reducer，join)。

4.1 Spark Shell

Spark提供了一个交互式shell - 一种以交互方式分析数据的强大工具。它以Scala或Python语言提供。Spark的主要抽象是一个名为Resilient Distributed Dataset(RDD)的分布式项目集合。可以从Hadoop输入格式(例如HDFS文件)或通过转换其他RDD来创建RDD。

4.2 打开Spark Shell

以下命令用于打开Spark shell。

$ spark-shell

4.3 创建简单的RDD

我们可以从文本文件中创建一个简单的RDD。使用以下命令创建简单的RDD。

scala> val inputfile = sc.textFile(“input.txt”)

Spark RDD API 提供了了一些转换和操作来处理RDD。

4.4 RDD 转换(RDD Transformations)

RDD转换返回指向新RDD的指针，并允许您在RDD之间创建依赖关系。依赖关系链中的每个RDD(依赖关系字符串)都有一个用于计算其数据的函数，并且具有指向其父RDD的指针(依赖关系)。

Spark很懒，所以除非你调用一些会触发作业创建和执行的转换或动作，否则不执行任何操作。请查看以下单词计数示例的片段。

因此，RDD转换不是一组数据，而是程序中的一个步骤(可能是唯一的步骤)，告诉Spark如何获取数据以及如何处理数据。

下面给出了RDD转换列表:

map(func):返回一个新的分布式数据集，它是通过函数func传递的每个元素而形成的。

filter(func)：返回通过选择func返回true的源元素形成的新数据集，也就是返回符合某种条件的数据集。

flatMap(func)：与map类似，但每个输入项可以映射到0个或更多输出项(因此func应该返回Seq而不是单个项)。

...

其他的这里不再一一列举，想要了解更多的，大家可以看下：Spark核心编程

4.5 RDD 操作

-reduce(func)：使用函数func(它接受两个参数并返回一个)来聚合数据集的元素。该函数应该是可交换的和关联的，以便可以并行正确计算。

collect()：以数组的形式返回数据集的所有元素。在过滤器或其他返回足够小的数据子集的操作之后，这通常很有用。

count()：返回数据集中的元素的个数。

first()：返回数据集的第一个元素(类似于take(1))。

take(n)：返回数据集的前n个元素的数组。

...

5 RDD 编程实例

5.1 单词计数实例

考虑单词计数示例 - 它计算出现在文档中的每个单词。将以下文本视为输入，并将其另存为input.txt文件。

people are not as beautiful as they look,

as they walk or as they talk.

they are only as beautiful as they love,

as they care as they share.

5.2 打开Spark-Shell

以下命令用于打开spark shell。通常，使用Scala构建spark。因此，Spark程序在Scala环境中运行。

$ spark-shell

如果Spark shell成功打开，那么您将找到以下输出。查看输出的最后一行“可用作sc的Spark上下文”意味着Spark容器会自动创建名为sc的 spark上下文对象。在开始程序的第一步之前，应该创建SparkContext对象。

5.3 创建一个RDD

首先，我们必须使用Spark-Scala API读取输入文件并创建RDD。

以下命令用于从给定位置读取文件。这里，使用inputfile的名称创建新的RDD。在textFile(“”)方法中作为参数给出的String是输入文件名的绝对路径。但是，如果仅给出文件名，则表示输入文件位于当前位置。

5.4 执行一个单词计数转换

我们的目的是统计一个文件的单词个数。首先创建一个flat map将文本的每行用空格分割。

然后使用 (map(word ⇒ (word, 1))将每个词作为key，value为1 ( = )

最后，通过添加类似键的值(reduceByKey(_ + _))来减少这些键.

以下命令用于执行字数统计逻辑。执行此操作后，您将找不到任何输出，因为这不是一个动作，这是一个转换; 指向一个新的RDD或告诉火花如何处理给定的数据)

val counts = inputfile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey(_+_);

5.5 当前的RDD

使用RDD时，如果您想了解当前的RDD，请使用以下命令。它将向您显示有关当前RDD及其调试依赖项的描述。

counts.toDebugString

5.6 缓存转换

可以使用persist()或cache()方法标记要保留的RDD。第一次在动作中计算它，它将保留在节点的内存中。使用以下命令将中间转换存储在内存中。

counts.cache()

5.7 执行操作

执行操作(如存储所有转换)会将结果导入文本文件。saveAsTextFile(“”)方法的String参数是输出文件夹的绝对路径。请尝试以下命令将输出保存在文本文件中。在以下示例中，'output'文件夹位于当前位置。

5.8 查看输出

我们输入命令查看统计结果：

bin/hdfs dfs -text /user/root/output/p*

5.9 查看Spark 任务和存储

在浏览器中使用以下URL：

参考资料