Kafka你真的了解吗？

一、Spark 介绍及生态

Spark 是 UC Berkeley AMP Lab 开源的通用分布式并行计算框架，目前已成为 Apache 软件基金会的顶级开源项目。至于为什么我们要学习 Spark，可以总结为下面三点：

1. Spark 相对于 hadoop 的优势

（1）高性能

Spark 具有 hadoop MR 所有的优点，hadoop MR 每次计算的中间结果都会存储到 HDFS 的磁盘上，而 Spark 的中间结果可以保存在内存，在内存中进行数据处理。

（2）高容错

• 基于“血统”(Lineage)的数据恢复：spark引入了弹性分布式数据集RDD的抽象，它是分布在一组节点中的只读的数据的集合，这些集合是弹性的且是相互依赖的，如果数据集中的一部分的数据发生丢失可以根据“血统”关系进行重建。
• CheckPoint容错：RDD计算时可以通过checkpoint进行容错，checkpoint有两种检测方式：通过冗余数据和日志记录更新操作。在RDD中的doCheckPoint方法相当于通过冗余数据来缓存数据，而“血统”是通过粗粒度的记录更新操作来实现容错的。CheckPoint容错是对血统检测进行的容错辅助，避免“血统”(Lineage)过长造成的容错成本过高。

（3）spark 的通用性

spark 是一个通用的大数据计算框架，相对于 hadoop 它提供了更丰富的使用场景。

spark 相对于 hadoop map reduce 两种操作还提供了更为丰富的操作，分为 action(collect,reduce,save…)和 transformations（map,union,join,filter…）,同时在各节点的通信模型中相对于 hadoop 的 shuffle 操作还有分区，控制中间结果存储，物化视图等。

2. spark 生态介绍

Spark 支持多种编程语言，包括 Java、Python、R 和 Scala。在计算资源调度层支持 local 模式，standalone 模式，yarn 模式以及 k8s 等。

同时 spark 有多组件的支持应用场景，在 spark core 的基础上提供了 spark Streaming，spark SQL，spark Mllib，spark R，GraphX 等组件。

spark Streaming 用于实时流计算，spark SQL 旨在将熟悉的 SQL 数据库查询与更复杂的基于算法的分析相结合，GraphX 用于图计算，spark Mllib 用于机器学习，spark R 用于对 R 语言的数据计算。

spark 支持多种的存储介质，在存储层 spark 支持从 hdfs，hive，aws 等读入和写出数据，也支持从 hbase，es 等大数据库中读入和写出数据，同时也支持从 mysql，pg 等关系型数据库中读入写出数据，在实时流计算在可以从 flume，kafka 等多种数据源获取数据并执行流式计算。

在数据格式上 spark 也支持的非常丰富，比如常见的 txt，json，csv 等格式。同时也支持 parquet，orc，avro 等格式，这几种格式在数据压缩和海量数据查询上优势也较为明显。

二、spark 原理及特点

1. spark core

Spark Core 是 Spark 的核心，其包含如下几个部分：

（1）spark 基础配置

sparkContext 是 spark 应用程序的入口，spark 应用程序的提交和执行离不开 sparkContext，它隐藏了网络通信，分布式部署，消息通信，存储体系，计算存储等，开发人员只需要通过 sparkContext 等 api 进行开发即可。

sparkRpc 基于 netty 实现，分为异步和同步两种方式。 事件总线 主要用于 sparkContext 组件间的交换，它属于监听者模式，采用异步调用。 度量系统 主要用于系统的运行监控。

（2）spark 存储系统

它用于管理 spark 运行中依赖的数据存储方式和存储位置，spark 的存储系统优先考虑在各节点以内存的方式存储数据，内存不足时将数据写入磁盘中，这也是 spark 计算性能高的重要原因。

我们可以灵活的控制数据存储在内存还是磁盘中，同时可以通过远程网络调用将结果输出到远程存储中，比如 hdfs，hbase 等。

（3）spark 调度系统

spark 调度系统主要由 DAGScheduler 和 TaskScheduler 组成。

DAGScheduler 主要是把一个 Job 根据 RDD 间的依赖关系，划分为多个 Stage，对于划分后的每个 Stage 都抽象为一个或多个 Task 组成的任务集，并交给 TaskScheduler 来进行进一步的任务调度。而 TaskScheduler 负责对每个具体的 Task 进行调度。

具体调度算法有 FIFO，FAIR：

• FIFO调度：先进先出，这是Spark默认的调度模式。
• FAIR调度：支持将作业分组到池中，并为每个池设置不同的调度权重，任务可以按照权重来决定执行顺序。

2. spark sql

spark sql 提供了基于 sql 的数据处理方法，使得分布式的数据集处理变的更加简单，这也是 spark 广泛使用的重要原因。

目前大数据相关计算引擎一个重要的评价指标就是：是否支持 sql，这样才会降低使用者的门槛。spark sql 提供了两种抽象的数据集合 DataFrame 和 DataSet。

DataFrame 是 spark Sql 对结构化数据的抽象，可以简单的理解为 spark 中的表，相比较于 RDD 多了数据的表结构信息(schema).DataFrame = Data + schema

RDD 是分布式对象集合，DataFrame 是分布式 Row 的集合，提供了比 RDD 更丰富的算子，同时提升了数据的执行效率。

DataSet 是数据的分布式集合 ，它具有 RDD 强类型的优点 和 Spark SQL 优化后执行的优点。DataSet 可以由 jvm 对象构建，然后使用 map，filter，flatmap 等操作函数操作。

3. spark streaming

这个模块主要是对流数据的处理，支持流数据的可伸缩和容错处理，可以与 Flume 和 Kafka 等已建立的数据源集成。Spark Streaming 的实现，也使用 RDD 抽象的概念，使得在为流数据编写应用程序时更为方便。

4. spark 特点

（1）spark 计算速度快

spark 将每个任务构建成 DAG 进行计算，内部的计算过程通过弹性式分布式数据集 RDD 在内存在进行计算，相比于 hadoop 的 mapreduce 效率提升了 100 倍。

（2）易于使用

spark 提供了大量的算子，开发只需调用相关 api 进行实现无法关注底层的实现原理。

通用的大数据解决方案

相较于以前离线任务采用 mapreduce 实现，实时任务采用 storm 实现，目前这些都可以通过 spark 来实现，降低来开发的成本。同时 spark 通过 spark SQL 降低了用户的学习使用门槛，还提供了机器学习，图计算引擎等。

（3）支持多种的资源管理模式

学习使用中可以采用 local 模型进行任务的调试，在正式环境中又提供了 standalone，yarn 等模式，方便用户选择合适的资源管理模式进行适配。

（4）社区支持

spark 生态圈丰富，迭代更新快，成为大数据领域必备的计算引擎。

三、spark 运行模式及集群角色

1. spark 运行模式

2. spark 集群角色

下图是 spark 的集群角色图，主要有集群管理节点 cluster manager，工作节点 worker，执行器 executor，驱动器 driver 和应用程序 application 五部分组成，下面详细说明每部分的特点。

（1）Cluster Manager

集群管理器，它存在于 Master 进程中，主要用来对应用程序申请的资源进行管理，根据其部署模式的不同，可以分为 local，standalone，yarn，mesos 等模式。

（2）worker

worker 是 spark 的工作节点，用于执行任务的提交，主要工作职责有下面四点：

• worker节点通过注册机向cluster manager汇报自身的cpu，内存等信息。
• worker 节点在spark master作用下创建并启用executor，executor是真正的计算单元。
• spark master将任务Task分配给worker节点上的executor并执行运用。
• worker节点同步资源信息和executor状态信息给cluster manager。

在 yarn 模式下运行 worker 节点一般指的是 NodeManager 节点，standalone 模式下运行一般指的是 slave 节点。

（3）executor

executor 是真正执行计算任务的组件，它是 application 运行在 worker 上的一个进程。这个进程负责 Task 的运行，它能够将数据保存在内存或磁盘存储中，也能够将结果数据返回给 Driver。

（4）Application

application 是 Spark API 编程的应用程序，它包括实现 Driver 功能的代码和在程序中各个 executor 上要执行的代码，一个 application 由多个 job 组成。其中应用程序的入口为用户所定义的 main 方法。

（5）Driver

驱动器节点，它是一个运行 Application 中 main 函数并创建 SparkContext 的进程。application 通过 Driver 和 Cluster Manager 及 executor 进行通讯。它可以运行在 application 节点上，也可以由 application 提交给 Cluster Manager，再由 Cluster Manager 安排 worker 进行运行。

Driver 节点也负责提交 Job，并将 Job 转化为 Task，在各个 Executor 进程间协调 Task 的调度。

（6）sparkContext

sparkContext 是整个 spark 应用程序最关键的一个对象，是 Spark 所有功能的主要入口点。核心作用是初始化 spark 应用程序所需要的组件，同时还负责向 master 程序进行注册等。

3. spark 其它核心概念

（1）RDD

它是 Spark 中最重要的一个概念，是弹性分布式数据集，是一种容错的、可以被并行操作的元素集合，是 Spark 对所有数据处理的一种基本抽象。可以通过一系列的算子对 rdd 进行操作，主要分为 Transformation 和 Action 两种操作。

• ‍‍‍‍‍Transformation(转换)：是对已有的RDD进行换行生成新的RDD，对于转换过程采用惰性计算机制，不会立即计算出结果。常用的方法有map，filter，flatmap等。
• Action(执行)：对已有对RDD对数据执行计算产生结果，并将结果返回Driver或者写入到外部存储中。常用到方法有reduce，collect，saveAsTextFile等。

（2）DAG

DAG 是一个有向无环图，在 Spark 中， 使用 DAG 来描述我们的计算逻辑。主要分为 DAG Scheduler 和 Task Scheduler。

图片出自：https://blog.csdn.net/newchitu/article/details/92796302

（3）DAG Scheduler

DAG Scheduler 是面向 stage 的高层级的调度器，DAG Scheduler 把 DAG 拆分为多个 Task，每组 Task 都是一个 stage，解析时是以 shuffle 为边界进行反向构建的，每当遇见一个 shuffle，spark 就会产生一个新的 stage，接着以 TaskSet 的形式提交给底层的调度器（task scheduler），每个 stage 封装成一个 TaskSet。DAG Scheduler 需要记录 RDD 被存入磁盘物化等动作，同时会需要 Task 寻找最优等调度逻辑，以及监控因 shuffle 跨节点输出导致的失败。

（4）Task Scheduler

Task Scheduler 负责每一个具体任务的执行。它的主要职责包括

• 任务集的调度管理；
• 状态结果跟踪；
• 物理资源调度管理；
• 任务执行；
• 获取结果。

（5）Job

job 是有多个 stage 构建的并行的计算任务，job 是由 spark 的 action 操作来触发的，在 spark 中一个 job 包含多个 RDD 以及作用在 RDD 的各种操作算子。

（6）stage

DAG Scheduler 会把 DAG 切割成多个相互依赖的 Stage，划分 Stage 的一个依据是 RDD 间的宽窄依赖。

在对 Job 中的所有操作划分 Stage 时，一般会按照倒序进行，即从 Action 开始，遇到窄依赖操作，则划分到同一个执行阶段，遇到宽依赖操作，则划分一个新的执行阶段，且新的阶段为之前阶段的 parent，然后依次类推递归执行。

child Stage 需要等待所有的 parent Stage 执行完之后才可以执行，这时 Stage 之间根据依赖关系构成了一个大粒度的 DAG。在一个 Stage 内，所有的操作以串行的 Pipeline 的方式，由一组 Task 完成计算。

（7）TaskSet Task

TaskSet 可以理解为一种任务，对应一个 stage，是 Task 组成的任务集。一个 TaskSet 中的所有 Task 没有 shuffle 依赖可以并行计算。

Task 是 spark 中最独立的计算单元，由 Driver Manager 发送到 executer 执行，通常情况一个 task 处理 spark RDD 一个 partition。Task 分为 ShuffleMapTask 和 ResultTask 两种，位于最后一个 Stage 的 Task 为 ResultTask，其他阶段的属于 ShuffleMapTask。

四、spark 作业运行流程

1. spark 作业运行流程

spark 应用程序以进程集合为单位在分布式集群上运行，通过 driver 程序的 main 方法创建 sparkContext 的对象与集群进行交互。具体运行流程如下：

• sparkContext向cluster Manager申请CPU，内存等计算资源。
• cluster Manager分配应用程序执行所需要的资源，在worker节点创建executor。
• sparkContext将程序代码和task任务发送到executor上进行执行，代码可以是编译成的jar包或者python文件等。接着sparkContext会收集结果到Driver端。

2. spark RDD 迭代过程

• sparkContext创建RDD对象，计算RDD间的依赖关系，并组成一个DAG有向无环图。
• DAGScheduler将DAG划分为多个stage，并将stage对应的TaskSet提交到集群的管理中心，stage的划分依据是RDD中的宽窄依赖，spark遇见宽依赖就会划分为一个stage，每个stage中包含来一个或多个task任务，避免多个stage之间消息传递产生的系统开销。
• taskScheduler 通过集群管理中心为每一个task申请资源并将task提交到worker的节点上进行执行。
• worker上的executor执行具体的任务。

3. yarn 资源管理器介绍

spark 程序一般是运行在集群上的，spark on yarn 是工作或生产上用的非常多的一种运行模式。

没有 yarn 模式前，每个分布式框架都要跑在一个集群上面，比如说 Hadoop 要跑在一个集群上，Spark 用集群的时候跑在 standalone 上。这样的话整个集群的资源的利用率低，且管理起来比较麻烦。

yarn 是分布式资源管理和任务管理管理，主要由 ResourceManager，NodeManager 和 ApplicationMaster 三个模块组成。

ResourceManager 主要负责集群的资源管理，监控和分配。对于所有的应用它有绝对的控制权和资源管理权限。

NodeManager 负责节点的维护，执行和监控 task 运行状况。会通过心跳的方式向 ResourceManager 汇报自己的资源使用情况。

yarn 资源管理器的每个节点都运行着一个 NodeManager，是 ResourceManager 的代理。如果主节点的 ResourceManager 宕机后，会连接 ResourceManager 的备用节点。

ApplicationMaster 负责具体应用程序的调度和资源的协调，它会与 ResourceManager 协商进行资源申请。ResourceManager 以 container 容器的形式将资源分配给 application 进行运行。同时负责任务的启停。

container 是资源的抽象，它封装着每个节点上的资源信息(cpu,内存，磁盘，网络等)，yarn 将任务分配到 container 上运行，同时该任务只能使用 container 描述的资源，达到各个任务间资源的隔离。

4. spark 程序在 yarn 上执行流程

spark on yarn 分为两种模式 yarn-client 模式，和 yarn—cluster 模式，一般线上采用的是 yarn-cluster 模式。

（1）yarn-client 模式

driver 在客户端本地执行，这种模式可以使得 spark application 和客户端进行交互，因为 driver 在客户端可以通过 webUI 访问 driver 的状态。同时 Driver 会与 yarn 集群中的 Executor 进行大量的通信，会造成客户机网卡流量的大量增加。

（2）yarn-cluster 模式

Yarn-Cluster 主要用于生产环境中，因为 Driver 运行在 Yarn 集群中某一台 NodeManager 中，每次提交任务的 Driver 所在的机器都是随机的，不会产生某一台机器网卡流量激增的现象，缺点是任务提交后不能看到日志。只能通过 yarn 查看日志。

下图是 yarn-cluster 运行模式：

client 向 yarn 提交应用程序，包含 ApplicationMaster 程序、启动 ApplicationMaster 的命令、需要在 Executor 中运行的程序等。

ApplicationMaster 程序启动 ApplicationMaster 的命令、需要在 Executor 中运行的程序等。

ApplicationMaster 向 ResourceManager 注册，这样用户可以直接通过 ResourceManage 查看应用程序的运行状态。

ApplicationMaster 申请到资源（也就是 Container）后，便与对应的 NodeManager 通信，启动 Task。

Task 向 ApplicationMaster 汇报运行的状态和进度，以让 ApplicationMaster 随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。

应用程序运行完成后，ApplicationMaster 向 ResourceManager 申请注销并关闭自己。

参考文献：

[1] Spark on Yarn 架构原理:

https://blog.csdn.net/lijingjingchn/article/details/85012470

[1] Spark on Yarn 详解:

https://www.cnblogs.com/bigdata1024/p/12116621.html

[1] Spark 任务提交方式和执行流程:

https://www.cnblogs.com/frankdeng/p/9301485.html

[1] Spark 容错机制:

https://www.cnblogs.com/cynchanpin/p/7163160.html

[1] spark 之 scheduler:

https://mp.weixin.qq.com/s/9g5e5WlmXUyQDXiU6PTGZA?token=1292183487&lang=zh_CN

[1] Spark 工作原理:

https://blog.csdn.net/qq_16681169/article/details/82432841

[1] Spark RDD:

https://www.cnblogs.com/zlslch/p/5942204.html

[1] Spark 基本概念快速入门:

https://www.leonlu.cc/profession/17-spark-terminology/

[1] spark 中 dag 的介绍:

https://blog.csdn.net/newchitu/article/details/92796302

[2] spark:

https://spark.apache.org/docs/3.0.0-preview/index.html