最简大数据Spark-2.1.0

wolearn 简书

0.0 前言

本文主要基于最新的Spark 2.1.0版本。阅读本文可以对Spark 2.1.0的学习过程，运行流程，关键组件，原理有所了解。文章有点长，你也可以直接阅读感兴趣的部分，但是还是建议全面了解。

1.0 简介

Spark是Apache软件基金会下的分布式系统开源项目。在官网中这样概况这个项目

Apache Spark is a fast and general engine for large-scale data processing.

说Spark是一个用于大规模数据处理的快速通用分布式引擎，到底有多快，请看下面这句

Lightning-fast cluster computing

快如闪电的集群计算。感觉官网用闪电来形容有点抽象，我总结了Spark大致有以下特点：

快。基于内存的批处理，在内存中比Hadoop的MR模型快100多倍，磁盘上也快了10倍。为什么这么快？Spark有先进的DAG（有向无环图）执行引擎还有其基于内存的计算方式有关。
易用。支持多语言Java，Scala，Python，R。同时能跟Hadoop生态圈很好的融合，比如Hive，HDFS，YARN。在近些年Hadoop已经慢慢成为大数据事实标准的大环境下，能抱住Hadoop的大腿，使其快速火起来。毕竟对于已经上了Hadoop的公司，切到一个如果完全不同的生态，即使这个生态再好，考虑到成本问题，公司也会犹豫。但是Spark完全不存在这个顾虑。
完整的技术栈。包括SQL查询、流式计算、机器学习和图算法组件在内的一揽子解决方案，让你一个框架搞定所有，难以拒绝的诱惑。

Spark可以很好的支持大数据中常见的三种场景批处理、交互式查询、流数据处理。但是其流处理只能达到秒级，如果有毫秒级别的需求，只能考虑Storm了。

2.0 怎么学习

首先看官网的教程

http://spark.apache.org/docs/latest/programming-guide.html

看官网的教程基本给出了Scala，Java，Python三种代码的例子，也就是说，你只要会其中一种就可以玩了。但是考虑到Spark源码是Scala写的，从学习源码的角度来看，用Scala是比较靠谱的。说下学习前的相关知识储备

服务端相关概念：如分布式，集群等
数据库相关概念：基本的数据库，数据库连接，Sql等
Hadoop生态：MR模型，Hive等
语言相关：推荐Scala，会Java，Python尤佳
Linux相关：因为Spark是部署在Linux上的，需要一些基本的操作。推荐安装Ubuntu，或CenterOS
IDE：Scala有自己的eclipse，也可以使用IDEA开发
其他：如Maven使用等

3.0 Begin

3.1 说在前面

本文主要梳理Spark学习中大的知识脉络，比较宏观的原理，希望有个整体的认知。至于具体的如相关API的调用，细小的概念，由于作者比较懒，请自行学习。

3.2 环境配置

可以从下面的地址下载到Spark的源码，或者编译后的包。

http://spark.apache.org/downloads.html

目前Spark中的1.X版本中最新的是1.6.3，而2.X版本中最新的是2.1.0。你可以安装Hadoop使用其中的分布式文件系统HDFS，也可以不安装只使用本地文件系统。解压完成后，建议将Spark加入Linux的环境变量，方便使用Spark的相关命令。Spark提供很好用的交互式工具，使用下面命令直接调用

spark-shell --master local[*]

除了shell的交互方式。Spark当然也可以运行独立应用程序。Java程序可以通过Maven来打包，Scala程序则通过sbt工具来打包，Python程序可以直接运行。用IDEA操作比较方便。Spark的提交命令如下

spark-submit

3.3 Main

3.3.0 Spark的生态

Spark的生态系统主要包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX 等件。

Spark Core：Spark 核心包，包含Spark的基础功能。
Spark SQL：Spark SQL允许开发人员以处理RDD的方式，直接处理外部的数据源，如HBase，Hive等。
MLlib：机器学习相关。
GraphX：图计算相关。

3.3.1 概念梳理

RDD（Resilient Distributed Dataset）：弹性分布式数据集，是最重要的一个概念，基本操作都是在和RDD打交道。RDD是Spark在分布式系统中抽象的一种高度受限的共享内存模型。个人感觉他的高度受限体现在他创建只能基于稳定的物理数据集合或RDD转化，而且创建后基本不能修改，只能通过转化操作，生成新的RDD完成变更。
DAG（Directed Acyclic Graph）：还记得之前说Spark为什么那么快吗？就是因为Spark有先进的DAG执行引擎。DAG主要是反映RDD之间的依赖关系。
Executor：Worker Node上的进程，负责运行任务，并储存数据。
Application：就是你编写的应用程序。
Job：作业。由多个阶段构成。包含多个RDD以及RDD上的操作。
Stage：阶段。由多个任务构成，是作业的基本调度单位。在Job和Task之间起到承上启下的作用。
Task：任务。Executor上的工作任务。
Cluster Manage：Spark集群获取资源的服务。Spark有自己的资源管理服务，但是也可以和YARN可靠的结合。Hadoop2.0后将资源协调抽离出来由YARN来负责，而Spark也可以使用YARN来协调资源。将各自为政的资源协调体系整合起来。这就是上面说的Spark可以很好的和Hadoop整合。
Driver Program：任务控制节点。由Driver为Executor上执行的task向Cluster Manage申请资源，资源包括CPU，内存等。

3.3.2 运行时流程

当一个Appliction被提交到Spark集群，Driver节点会先创建一个SparkContext。由SparkContext向Cluster Manager申请资源。
Cluster Manager在Work Node上启动Executor进程，等待执行Task，并为其分配运行需要的资源，同时随时监测其资源运行情况。
Executor拿到资源并运行起来之后，向Driver注册，并告知其可以执行任务。
Spark Application将任务抽象成RDD并构建DAG，并由DAGSchedule将DAG解析成Stage，并计算Stage的依赖关系，提交给TaskSchedule。
TaskSchedule将任务分配到注册到Driver的Executor上，执行起来。
Executor将执行结果告之TaskSchedule，TaskSchedule将结果告之DAGSchedule，然后由SparkContent通知Cluster Manager任务运行完毕并释放资源。

3.4 理解RDD

对RDD的理解主要对照官方文档2.1.0的理解梳理。

3.4.1 综述

The main abstraction Spark provides is a resilient distributed dataset (RDD), which is a collection of elements partitioned across the nodes of the cluster that can be operated on in parallel. RDDs are created by starting with a file in the Hadoop file system (or any other Hadoop-supported file system), or an existing Scala collection in the driver program, and transforming it. Users may also ask Spark to persist an RDD in memory, allowing it to be reused efficiently across parallel operations. Finally, RDDs automatically recover from node failures.

Spark提供弹性分布式数据集（RDD）作为主要抽象，它可以提供并行的操作在集群的节点之间。（RDD创建）可以通过从Hadoop文件系统（或任何其他Hadoop支持的文件系统）中的文件或驱动程序中的现有Scala集合，或者通过RDD的转化操作得到。（RDD的持久化）用户还可以要求Spark 在内存中保留 RDD，从而在并行操作中有效地重用RDD。（RDD基于血缘关系的高容错）最后，RDD自动从节点故障中恢复。

3.4.2 RDD来源

There are two ways to create RDDs: parallelizing an existing collection in your driver program, or referencing a dataset in an external storage system, such as a shared filesystem, HDFS, HBase, or any data source offering a Hadoop InputFormat.

创建RDD的两种方法：并行化（即通过parallelize方法），或者在外部存储系统（如共享文件系统，HDFS，HBase或提供Hadoop InputFormat的任何数据源）中引用数据集。

3.4.3 RDD操作

RDDs support two types of operations: transformations, which create a new dataset from an existing one, and actions, which return a value to the driver program after running a computation on the dataset.
All transformations in Spark are lazy, in that they do not compute their results right away. Instead, they just remember the transformations applied to some base dataset (e.g. a file). The transformations are only computed when an action requires a result to be returned to the driver program. This design enables Spark to run more efficiently.

RDDS支持两种类型的操作：transformations(转化)，从现有的RDD创建一个新的RDD。actions(行动)，其上运行的数据集计算后获取值返回驱动程序。
Spark中的所有transformations操作都是懒操作，因为它们不会马上计算它们的结果。他们只记住应用于某些基本数据集（例如文件）的转换。只有当某个actions操作执行的时候，transformations操作才会真正执行。此设计使Spark能够更高效地运行。

4.0 Spark SQL

4.1 SQL

Spark SQL是用于结构化数据处理的Spark模块。

Spark SQL的一个用途是执行SQL查询。Spark SQL也可用于从Hive中读取数据。当编程语言运行SQL时，结果将以Dataset / DataFrame的形式返回。还可以使用命令行或JDBC / ODBC与SQL界面进行交互。

4.2 Datasets

Dataset是分布式数据集。Dataset是Spark 1.6中开始添加的新功能，它使RDD具备了强类型，强大的lambda函数等功能，同时具有Spark SQL优化的执行引擎的优点。Dataset可以从JVM对象中创建，然后使用功能性的转换（操作map，flatMap，filter等等）。数据集API可用于Scala和Java。Python不支持Dataset API。

Dataset跟RDD类似，但是他不使用Java的序列化或者Kryo，他有专门的Encoder（编码器）串行化对象用于网络的传输和处理。虽然Encoder和标准序列化都将负责将对象转换成字节，但是Encoder使用了一种格式去构造代码，使得Spark可以进行许多操作（如map，filter），而无需经过字节反序列化到对象的过程。

4.3 DataFrames

DataFrame是一种被组织进有名称的列的Dataset（很拗口，有没有？看接下来比较好懂）。他在概念上等同于关系型数据库中的表，但是适用面更广。DataFrames可以从各种各样的源构建，例如：结构化数据文件，Hive中的表，外部数据库或现有RDD。

而官方文档的大部分篇幅都是在讲两件事：

构造一个DataFrames。从文件中构建，从结构化数据如json、Hive中的表中构建，RDD的转换操作等
操作DataFrames。

4.4 个人理解

Spark SQL大部分时候其实就是抽象出一种数据结构DataFrame，然后通过操作DataFrame的形式间接操作结构化数据。

5.0 Spark Streaming

5.1 概述

Spark Streaming 是Spark核心API的扩展，能够处理高扩展，高吞吐，高容错的实时数据流。能够接收多种数据来源，如Kafka, Flume, Kinesis, 或者 TCP sockets，并且能使用复杂的算法表达式如，map, reduce, join 和 window。最后，处理过的数据可以推送到文件系统，数据库，和实时展示的图表。