目录

一，MapReduce

1，简介

2，原理

2.1 基本概念

2.2 程序执行过程

2.3 作业运行模式

二，Spark

1，简介

1.1 背景

1.2 概念

1.3 特点

2，原理

2.1 编程模型

2.2 运行模式

2.3 运行过程

2.4 DAG任务规划与调度

---

文章内容来自：南京大学 / 星环科技课程，大数据理论与实践课程Ⅰ

对细节部分引用其他网络资源进行补充。

一，MapReduce

1，简介

MR是面向离线批处理的分布式计算框架

核心思想：分而治之，并行计算。移动计算，非移动数据；

适用场景

• 数据统计，如网站的PV、UV统计
• 搜索引擎构建索引
• 海量数据查询
• 复杂数据分析算法实现

不适用场景

• OLAP：要求毫秒或秒级返回结果
• 流计算：输入数据集是动态的，而MapReduce是静态的
• DAG计算
• －多个任务之间存在依赖关系，后一个的输入是前一个的输出，构成DAG有向无环图
• －MapReduce很难避免Suffle，造成大量磁盘IO，导致性能较为低下

2，原理

2.1 基本概念

1，Job & Task（作业与任务）

• 作业是客户端请求执行的一个工作单元。包括输入数据、MapReduce程序、配置信息
• 任务是将作业分解后得到的细分工作单元。分为Map任务和Reduce任务

2，Split（切片）

• 输入数据被划分成等长的小数据块，称为输入切片（Input Split），简称切片
• Split是逻辑概念，仅包含元数据信息，如数据的起始位置、长度、所在节点等
• 每个Split交给一个Map任务处理，Split的数量决定Map任务的数量
• Split大小默认等于HDFS Block大小，Split的划分方式由程序设定，Split与HDFS Block没有严格的对应关系。Split越小，Map任务越多，并发度越高，但开销也越大；Split越大，任务越少，并发度降低

3，Map阶段（映射）

• 由若干Map任务组成，任务数量由Split数量决定
• 输入：Split切片（key-value） 。输出：中间计算结果（key-value）

4，Reduce阶段（化简）

• 由若干Reduce任务组成，任务数量由程序指定
• 输入：Map阶段输出的中间结果（key-value）。输出：最终结果（key-value） 

5，Shuffle阶段（混洗）

• Shuffle是Map和Reduce之间的强依赖关系（Shuffle依赖）导致的，即每个Reduce的输入依赖于所有Map的输出
• Map和Reduce阶段的中间环节（虚拟阶段），分为Map端Shuffle和Reduce端Shuffle
• 包括Partition（分区）、Sort（排序）、Spill（溢写）、Merge（合并）和Fetch（抓取）等工作

Partition（分区）

• Reduce任务数量决定了Partition数量，Partition编号 = Reduce任务编号
• 利用“哈希取模”对Map输出数据分区，即Partition编号 = key hashcode % reduce task num（%为取模）
• Partition为具有相同编号的Reduce任务供数

哈希取模的作用

• 数据划分：将一个数据集随机分成若干个子集（Hash函数选择不当可能造成数据倾斜）
• 数据聚合：将Key相同的数据聚合在一起

避免和减少Shuffle是MapReduce程序调优的关键

2.2 程序执行过程

1，MR执行过程

 

2，shuffle详解

注意：在溢写之前需要先进行排序（便于后续归并排序），MapTask结束后仍需要通过归并排序将所有溢写文件合并为一个文件。

Map端

1，Map任务将中间结果写入环形内存缓冲区Buffer（默认100M）；

2，当Buffer的数据量达到阈值（默认80%）时，对缓冲区内数据进行分区（Partition）和排序（Sort）。 先按“key hashcode % reduce task num”对数据进行分区，分区内再按key排序。然后将数据溢写（Spill）到磁盘的一个临时文件中。如果在溢写过程中，剩余20%的空间又被耗尽，这时就会触发panding，等80%空间腾出来之后再继续写；

3，Map任务结束前，将多个临时文件合并（Merge）为一个Map输出文件，文件内数据先分区后排序

Reduce端

1，Reduce任务从多个Map输出文件中抓取（Fetch）属于自己的分区数据（Partition编号=Reduce任务编号）

2，对抓取到的分区数据做归并排序，生成一个Reduce输入文件（文件内数据按key排序）

• 如果内存缓冲区够大，就直接在内存中完成归并排序，然后落盘
• 如果内存缓冲区不够，先将分区数据写到相应的文件中，再通过归并排序合并为一个大文件

关于环形缓冲区的介绍可以参考这里@大数据架构师Evan【设计思想赏析-MapReduce环形缓冲区】

2.3 作业运行模式

1，JobTracker/TaskTracker模式（Hadoop 1.X）

 

JobTracker节点（Master）

• 调度任务在TaskTracker上运行
• 若任务失败，指定新TaskTracker重新运行

TaskTracker节点（Slave）

• 执行任务，发送进度报告

存在的问题

• JobTracker存在单点故障
• JobTracker负载太重（上限4k节点）
• JobTracker缺少对资源的全面管理
• TaskTracker对资源的描述过于简单
• 源码难于理解

 

2，YARN模式（Hadoop 2.X ）

二，Spark

1，简介

1.1 背景

MapReduce有较大的局限性

• 仅支持Map、Reduce两种语义操作，划分为两个阶段（模型较为粗糙）
• 执行效率低，时间开销大（很难避免Shuffle）
• 主要用于大规模离线批处理
• 不适合迭代计算、在线分析、实时流处理等场景

计算框架种类多，选型难，学习成本高

• 批处理：MapReduce
• 流处理：Storm、Flink
• 在线分析：Impala、Presto
• 机器学习：Mahout

统一计算框架，简化技术选型，降低学习成本

• 在统一框架下，实现离线批处理、流处理、在线分析和机器学习

1.2 概念

由加州大学伯克利分校的AMP实验室开源

  

高性能的分布式通用计算引擎

• Spark Core：核心计算框架
• Spark SQL：结构化数据查询
• Spark Streaming：实时流处理
• Spark MLib：机器学习
• Spark GraphX：图计算

具有高吞吐、低延时、通用易扩展、高容错等特点

采用Scala语言开发

提供多种运行模式

1.3 特点

计算高效

• 语义操作多样，模型设计精细
• 利用RDD内存计算和Cache缓存机制，支持迭代计算和数据共享，减少数据读取的IO开销
• 利用DAG引擎，减少中间计算结果写入HDFS的开销
• 利用多线程池模型，减少任务启动开销，避免Shuffle中不必要的排序和磁盘IO操作

  

通用易用

• 适用于批处理、流处理、在线分析、机器学习等场景
• 提供了丰富的开发API，支持Scala、Java、Python、R等

运行模式多样

• Local模式
• Standalone模式
• YARN/Mesos模式

2，原理

2.1 编程模型

1，RDD（Resilient Distributed Datesets） 弹性分布式数据集

RDD相当于Table，由分布在集群中的多个Partition组成

Partition（分区）

• 分布在集群的不同节点中
• 只读数据集
• 通过转换操作来构造
• 失效后自动重构（弹性）
• 存储在内存或磁盘中

Spark基于RDD进行计算

 

2，RDD操作（Operator）

Transformation（转换）

• 将Scala集合或Hadoop输入数据构造成一个新RDD
• 通过已有的RDD产生新RDD
• 惰性执行：只记录转换关系，不触发计算
• 例如：map、filter、flatmap、union、distinct、sortbykey

Action（动作）

• 通过RDD计算得到结果或者落盘
• 真正触发计算
• 例如：first、count、collect、foreach、saveAsTextFile

以rdd1.map(_,+1).saveAsTextFile(“hdfs://node01:9000”)为例

3，RDD依赖（Dependency）

窄依赖（Narrow Dependency）

• 每个父RDD分区只能为一个子RDD分区供数，
• 子分区所依赖的父分区集合之间没有交集
• 子RDD分区数据丢失或损坏，从其依赖的父RDD分区重新计算即可，无需Shuffle
• 例如：map、filter、union

宽依赖（Wide/Shuffle Dependency）

• 每个父RDD分区为所有子RDD分区供数
• 子RDD区数据丢失或损坏，从所有父RDD分区重新计算，必须Shuffle
• 相对于窄依赖，宽依赖付出的代价要高很多，尽量避免使用
• 例如：groupByKey、reduceByKey、sortByKey

 

4，示例：WordCount

val rdd1 = sc.textFile(“hdfs://node01:9000/data/in”)

val rdd2 = rdd1.flatMap(_.split(“\t”))

val rdd3 = rdd2.map((_,1))

val rdd4 = rdd3.reduceByKey((_+_))

rdd4.saveAsTextFile(“hdfs://node01:9000/data/out”)

 

 

2.2 运行模式

1，抽象模式

Driver

• 每个Spark作业启动一个Driver，每个Driver创建一个SparkContext
• 负责解析Spark程序、划分Stage、调度任务到Executor上执行

SparkContext

• 负责加载配置信息，初始化运行环境，创建DAGScheduler和TaskScheduler
• DAGScheduler：根据任务依赖建立DAG、根据宽依赖划分Stage、提交TaskSet
• TaskScheduler：任务调度和监管

Executor

• 负责执行Driver分发的任务，一个节点可以启动多个Executor，每个Executor通过多线程运行多个任务

Task

• Spark运行的基本单位，一个Task负责处理若干RDD分区的计算逻辑

2，Local模式

单机运行，通常用于测试

Spark程序以多线程方式直接运行在本地

3，Standalone模式

Spark集群独立运行，不依赖于第三方资源管理系统，如YARN、Mesos

采用Master/Slave架构

• Master统一管理集群
• Driver在Worker中运行

ZooKeeper负责Master HA，避免单点故障

适用于集群规模和数据量都不大的情况

4，YARN模式

YARN-Client模式：适用于交互和调试

YARN-Cluster模式：适用于生产环境 

区别在于Driver位置。Driver放在Client主要便查看日志，便于调试。

  

 

2.3 运行过程

1，生成逻辑计划

 

2，生成物理计划

3，任务调度与执行

 

2.4 DAG任务规划与调度

DAG（Directed Acyclic Graph）

• 有向无环图DAG：一个有向图无法从任意顶点出发经过若干条边回到该点
• 受制于某些任务必须比另一些任务早执行的约束，Spark程序的内部执行逻辑可由DAG描述，节点代表任务，边代表任务间的依赖约束

DAGScheduler

• 根据任务的依赖关系建立DAG
• 根据依赖关系是否为宽依赖，即是否存在Shuffle，将DAG划分为不同的阶段（Stage）
• 将各阶段Task组成的TaskSet提交给TaskScheduler

TaskScheduler

• 负责任务调度
• 重新提交失败的Task
• 为执行速度慢的Task启动备用Task