Hadoop与Spark大数据

海量数据的存储问题很早就已经出现了，一些行业或者部门因为历史的积累，数据量也达到了一定的级别。很早以前，当一台电脑无法存储这么庞大的数据时，采用的解决方案是使用NFS(网络文件系统)将数据分开存储。但是这种方法无法充分利用多台计算机同时进行分析数据。

一个实际的需求场景：日志分析

日志分析是对日志中的每一个用户的流量进行汇总求和。对于一个日志文件，如果只有这么几行数据，我们一般会采用这样的处理方式

• 读取一行日志
• 抽取手机号和流量字段
• 累加到HashMap中
• 遍历输出结果

如果数据量变得很大呢，比如一个日志文件里面有几个GB数据，仍然一行一行去读，那么就会因为磁盘的IO瓶颈导致效率太低，速度太慢。如果一次性加载到内存，那么就会因为单台计算机的内存空间有限而导致内存溢出。如果将中间结果全部缓存到HashMap中，那么也会因为单台计算机的内存空间有限而导致内存溢出。可以选择采用多线程处理，但是依然无法改变资源瓶颈的现实，因为一台计算器的CPU资源，内存资源，磁盘IO瓶颈是定，创建再多的线程也无法改变这个现实。

解决思路一：纵向扩展

升级硬件，提高单机性能(增加内存，增强CPU、用更高性能的磁盘(如固态硬盘)，比如可以购买IBM的高端服务器。这种方法简单易行，但单台计算机的扩展空间有限，CPU、内存、磁盘再怎么扩展也是有限的，无法无限扩展。除此之外，成本较高，高端服务器常常非常昂贵，几百万甚至上千万一台，一般的小公司承受不起这样高昂的成本。

解决思路二：横向扩展:

用多台节点分布式集群处理 ，通过增加节点数量提高处理能力，这里说的节点指的就是一台计算机。其思想为任务分摊，通过协作来实现单节点无法实现的任务，且成本相对低(可采用普通机器) 、易于线性扩展。但是其系统复杂度增加，我们要将我们的web应用部署到每一个节点上面，而多个节点协同工作时就要考虑：如何调度资源、任务如何监控、中间结果如何调度、系统如何容错、如何实现众多节点间的协调等问题。

Hadoop

Hadoop是一个开源的可运行于大规模集群上的分布式文件系统和运行处理基础框架。其擅长于在廉价机器搭建的集群上进行海量数据(结构化与非结构化)的存储与离线处理。Hadoop就是一门用来处理大数据的技术，就是用来解决上述提到的分布式计算里面的5个技术难题的。

Hadoop 由 Apache Software Foundation 公司于 2005 年秋天作为 Lucene 的子 项目 Nutch 的一部分正式引入。它受到最先由 Google Lab 开发的 MapReduce 和 Google File System 的启发。2006 年 3 月份，MapReduce 和 Nutch Distributed File System (NDFS) 分别被纳入称为 Hadoop 的项目中。　　Hadoop 是最受欢迎的在 Internet 上对搜索关键字进行内容分类的工具，但它也可以解决许多要求极大伸缩性的问题。例如，如果您要 grep 一个 10TB 的巨型文件，会出现什么情况？在传统的系统上，这将需要很长的时间。但是 Hadoop 在设计时就考虑到这些问题，因此能大大提高效率。

Hadoop的生态坏境

Hadoop是一个能够对大量数据进行分布式处理的软件框架。具有可靠、高效、可伸缩的特点。Hadoop的核心是HDFS和Mapreduce，hadoop2.0还包括YARN。

HDFS（Hadoop分布式文件系统）

HDFS（Hadoop Distributed File System）源自于Google的GFS论文，发表于2003年10月，HDFS是GFS克隆版。是Hadoop体系中数据存储管理的基础。它是一个高度容错的系统，能检测和应对硬件故障，用于在低成本的通用硬件上运行。HDFS简化了文件的一致性模型，通过流式数据访问，提供高吞吐量应用程序数据访问功能，适合带有大型数据集的应用程序。

• Client：切分文件；访问HDFS；与NameNode交互，获取文件位置信息；与DataNode交互，读取和写入数据。
• NameNode：Master节点，在hadoop1.X中只有一个，管理HDFS的名称空间和数据块映射信息，配置副本策略，处理客户端请求。
• DataNode：Slave节点，存储实际的数据，汇报存储信息给NameNode。
• Secondary NameNode：辅助NameNode，分担其工作量；定期合并fsimage和fsedits，推送给NameNode；紧急情况下，可辅助恢复NameNode，但Secondary NameNode并非NameNode的热备。

YARN是一种 Hadoop 资源管理器，它是一个通用资源管理系统，可为上层应用提供统一的资源管理和调度，它的引入为集群在利用率、资源统一管理和数据共享等方面带来了巨大好处。
#####Hadoop YARN产生背景
直接源于MRv1在几个方面的缺陷：

• 扩展性受限
• 单点故障
• 难以支持MR之外的计算

除此之外，多计算框架各自为战，数据共享困难。比如：MapReduce（离线计算框架）、Storm（实时计算框架）、Spark（内存计算框架）。
#####Hadoop YARN基本架构

ResourceManager

• 处理客户端请求
• 启动/监控ApplicationMaster
• 监控NodeManager
• 资源分配与调度

NodeManager

• 单个节点上的资源管理
• 处理来自ResourceManager的命令
• 处理来自ApplicationMaster的命令

ApplicationMaster

• 数据切分
• 为应用程序申请资源，并分配给内部任务
• 任务监控与容错

Hadoop YARN调度框架

Hadoop YARN采用双层调度框架。首先，RM将资源分配给AM。其次，AM再将资源进一步分配给各个Task。YARN与“all or nothing”策略不同（Apache Mesos），其遵循基于资源预留的调度策略，当资源不够时，会为Task预留，直到资源充足。

运行在YARN上的计算框架

• 离线计算框架：MapReduce
• DAG计算框架：Tez
• 流式计算框架：Storm
• 内存计算框架：Spark
• 图计算框架：Giraph、GraphLib

Hadoop MapReduce是google MapReduce 克隆版。

MapReduce是一种计算模型，用以进行大数据量的计算。其中Map对数据集上的独立元素进行指定的操作，生成键-值对形式中间结果。Reduce则对中间结果中相同“键”的所有“值”进行规约，以得到最终结果。MapReduce这样的功能划分，非常适合在大量计算机组成的分布式并行环境里进行数据处理。

• JobTracker：Master节点，只有一个，管理所有作业，作业/任务的监控、错误处理等；将任务分解成一系列任务，并分派给TaskTracker。
• TaskTracker：Slave节点，运行Map Task和Reduce Task；并与JobTracker交互，汇报任务状态。
• Map Task：解析每条数据记录，传递给用户编写的map(),并执行，将输出结果写入本地磁盘(如果为map-only作业，直接写入HDFS)。
• Reducer Task：从Map Task的执行结果中，远程读取输入数据，对数据进行排序，将数据按照分组传递给用户编写的reduce函数执行。

Hive（基于Hadoop的数据仓库）

Hive由facebook开源，最初用于解决海量结构化的日志数据统计问题。Hive定义了一种类似SQL的查询语言(HQL),将SQL转化为MapReduce任务在Hadoop上执行，通常用于离线分析。

HBase（分布式列存数据库）

Hbae源自Google的Bigtable论文，发表于2006年11月，HBase是Google Bigtable克隆版。HBase是一个针对结构化数据的可伸缩、高可靠、高性能、分布式和面向列的动态模式数据库。和传统关系数据库不同，HBase采用了BigTable的数据模型：增强的稀疏排序映射表（Key/Value），其中，键由行关键字、列关键字和时间戳构成。HBase提供了对大规模数据的随机、实时读写访问，同时，HBase中保存的数据可以使用MapReduce来处理，它将数据存储和并行计算完美地结合在一起。

HBase和Hive在大数据架构中处在不同位置，HBase主要解决实时数据查询问题，Hive主要解决数据处理和计算问题，一般是配合使用。

Zookeeper（分布式协作服务）

源自Google的Chubby论文，发表于2006年11月，Zookeeper是Chubby克隆版
解决分布式环境下的数据管理问题：统一命名，状态同步，集群管理，配置同步等。

Sqoop（数据同步工具）

Sqoop是SQL-to-Hadoop的缩写，主要用于传统数据库和Hadoop之前传输数据。数据的导入和导出本质上是Mapreduce程序，充分利用了MR的并行化和容错性。

Pig（基于Hadoop的数据流系统）

由yahoo!开源，设计动机是提供一种基于MapReduce的ad-hoc(计算在query时发生)数据分析工具。其定义了一种数据流语言—Pig Latin，将脚本转换为MapReduce任务在Hadoop上执行，通常用于进行离线分析。

Spark

Spark是UC Berkeley AMPLab开发的是一种计算框架，分布式资源工作交由集群管理软件（Mesos、YARN） 。

###Spark的特点

• 先进架构
• Spark采用Scala语言编写，底层采用了actor model的akka作为通讯框架，代码十分简洁高效。
• 基于DAG图的执行引擎，减少多次计算之间中间结果写到Hdfs的开销。
• 建立在统一抽象的RDD（分布式内存抽象）之上,使得它可以以基本一致的方式应对不同的大数据处理场景。
• 高效
• 提供Cache机制来支持需要反复迭代的计算或者多次数据共享，减少数据读取的IO开销。
• 与Hadoop的MapReduce相比，Spark基于内存的运算比MR要快100倍；而基于硬盘的运算也要快10倍！
• 易用
• Spark提供广泛的数据集操作类型（20+种），不像Hadoop只提供了Map和Reduce两种操作。
• Spark支持Java，Python和Scala API，支持交互式的Python和Scala的shell。
• 提供整体解决方案
• 以其RDD模型的强大表现能力，逐渐形成了一套自己的生态圈，提供了full-stack的解决方案。
• 主要包括Spark内存中批处理，Spark SQL交互式查询，Spark Streaming流式计算， GraphX和MLlib提供的常用图计算和机器学习算法。
• 与Hadoop无缝连接
• Spark可以使用YARN作为它的集群管理器
• 读取HDFS,HBase等一切Hadoop的数据

Spark整体架构

Spark提供了多种高级工具，如： Shark SQL应用于即席查询（Ad-hoc query）、Spark Streaming应用于流式计算、 MLlib应用于机器学习、GraphX应用于图处理。Spark还可以基于自带的standalone集群管理器独立运行，也可以部署在Apache Mesos 和 Hadoop YARN 等集群管理器上运行。Spark可以访问存储在HDFS、 Hbase、Cassandra、Amazon S3、本地文件系统等等上的数据，Spark支持文本文件，序列文件，以及任何Hadoop的InputFormat。

Spark核心概念

Spark任务提供多层分解的概念，Spark组件将用户的应用程序分解为内部执行任务并提供执行容器，资源管理为spark组件提供资源管理和调度。

Spark任务

• 应用程序：由一个driver program和多个job构成
• job：由多个stage组成
• stage：对应一个taskset
• taskset：对应一组关联的相互之间没有shuffle依赖关系的task组成。
• task：任务最小的工作单元

• Spark组件
• Driver Program (驱动程序) ：Spark 的核心组件
• 构建SparkContext(Spark应用的入口,它负责和整个集群的交互，创建需要的变量，还包含集群的配置信息等)
• 将用户提交的job转换为DAG图(类似数据处理的流程图)
• 根据策略将DAG图划分为多个stage，根据分区从而生成一系列tasks
• 根据tasks要求向资源管理器申请资源
• 提交任务并检测任务状态
• Executor
• 真正执行task的单元，一个Worker Node上可以有多个Executor
• 资资源管理组件
• YARN（通用）：Master/Slave结构
• Resource Manager(RM)：全局的资源管理器，负责系统的资源管理和分配
• Node Manager(NM)：每个节点上的资源和任务管理器
• Application Master(AM)：每个应用程序都有一个，负责任务的调度和监视，并与RM调度器协商为任务获取资源。
• Standalone（Spark自带）：Master/Slave结构
• Master：类似于YARN的RM
• Worker：类似于YARN的NM

Spark端到端流程

Spark的Driver Program (简称Driver)包含用户的应用程序，Driver完成task的解析和生成。

1. Driver向Cluster Manager（集群资源管理器）申请运行task需要的资源。
2. 集群资源管理器为task分配满足要求的节点，并在节点按照要求创建Executor
3. 创建的Executor向Driver注册。
4. Driver将Spark应用程序的代码和文件传送给分配的Executor
5. Executor运行task，运行完之后将结果返回给Driver或者写入HDFS或其他介质。

Spark流程

Spark on Standalone

Master和Worker是Standalone的角色，Driver和Executor是Spark的角色。Master负责分配资源，分配Driver和Executor，让Worker启动Driver和Executor，只管理到Executor层，不涉及任务。Driver负责生成task，并与Executor通信，进行任务的调度和结果跟踪，不涉及资源。

Driver运行在Worker

1. 客户端把作业发布到Master
2. Master让一个Worker启动Driver，并将作业推送给Driver
3. Driver进程生成一系列task
4. Driver向Master申请资源
5. Master让调度的Worker启动Exeuctor
6. Exeuctor启动后向Driver注册。
7. Driver将task调度到Exeuctor执行。
8. Executor执行结果写入文件或返回Driver

1. 客户端启动后直接运行用户程序，启动Driver
2. Driver进程生成一系列task
3. Driver向Master申请资源
4. Master让调度的Worker启动Exeuctor
5. Exeuctor启动后向Driver注册。
6. Driver将task调度到Exeuctor执行。
7. Executor执行结果写入文件或返回Driver

Spark的核心组件

Spark的核心组件包括RDD、Scheduler、Storage、Shuffle四部分：

• RDD是Spark最核心最精髓的部分，Spark将所有数据都抽象成RDD。
• Scheduler是Spark的调度机制，分为DAGScheduler和TaskScheduler。
• Storage模块主要管理缓存后的RDD、shuffle中间结果数据和broadcast数据
• Shuffle分为Hash方式和Sort方式，两种方式的shuffle中间数据都写本地盘

RDD（Resilient Distributed Datasets）

RDD是弹性分布式数据集，是只读的分区记录集合。每个RDD有5个主要的属性：

• 一组分片(Partition)：数据集的最基本组成单位
• 一个计算每个分片的函数：对于给定的数据集，需要做哪些计算
• 依赖（Dependencies）：RDD的依赖关系，描述了RDD之间的lineage
• preferredLocations（可选）：对于data partition的位置偏好
• partitioner（可选）：对于计算出来的数据结果如何分发

作用于RDD上的Operation分为转换(transformantion)和动作(action)。 Spark中的所有“转换”都是惰性的，在执行“转换”操作，并不会提交Job，只有在执行“动作”操作，所有operation才会被提交到cluster中真正的被执行。这样可以大大提升系统的性能。

• 转换：从现有的数据集创建一个新的数据集即数据集中的内容会发生更改，由数据集A转换成为数据集B
• 动作：在数据集上运行计算后，返回一个值给驱动程序。 即数据集中的内容会被归约为一个具体的数值（Scala标量、集合类型的数据或存储）。

RDD拥有的操作比MR丰富的多，不仅仅包括Map、Reduce操作，还包括右图的filter、sort、join、save、count等操作，所以Spark比MR更容易方便完成更复杂的任务。

RDD持久化

默认情况下，每一个转换过的RDD都会在它之上执行一个动作时被重新计算。如果RDD只被使用一次或者很少次，不需要持久化。如果RDD被重复使用或者计算其代价很高，才考虑持久化。另外，shuffle后生成的RDD尽量持久化，因为shuffle代价太高。RDD被缓存后，Spark将会在集群中，保存相关元数据，下次查询这个RDD时，它将能更快速访问，不需要计算。如果持久化无谓的RDD，会浪费内存（或硬盘）空间，反而降低系统整体性能

RDD依赖关系

RDD只能基于在稳定物理存储中的数据集和其他已有的RDD上执行确定性操作来创建。能从其他RDD通过确定操作创建新的RDD的原因是RDD含有从其他RDD衍生（即计算）出本RDD的相关信息（即Lineage）。Dependency代表了RDD之间的依赖关系，即血缘（Lineage），分为窄依赖和宽依赖：

• 窄依赖：一个父RDD最多被一个子RDD用在一个集群节点上管道式执行。比如map、filter、union等
• 宽依赖：子RDD的分区依赖于父RDD的所有分区，这是因为shuffle类操作要求所有父分区可用。比如groupByKey、reduceByKey、 sort、partitionBy等

根据RDD依赖关系的不同，Spark将每一个job分为不同的stage，stage之间的依赖关系形成了DAG图。对于窄依赖，Spark将其尽量划分在同一个stage中，因为它们可以进行流水线计算，而宽依赖往往意味着shuffle操作，这也是Spark划分stage的主要边界。

宽/窄依赖的概念不止用在stage划分中，对容错也很有用。若Transformation操作中间发生计算失败，如果运算是窄依赖，只要把丢失的父RDD分区重算即可，跟其他节点没有依赖，这样可以大大加快场景恢复的开销，但如果运算是宽依赖，则需要父RDD的所有分区都存在， 重算代价就较高。当Lineage特别长时或者有宽依赖时，主动调用 checkpoint把当前数据写入稳定存储，作为检查点。

Scheduler

Scheduler模块作为Spark最核心的模块之一，充分体现了Spark与MapReduce的不同之处，体现了Spark DAG思想的精巧和设计的优雅。Scheduler模块分为两大主要部分，DAGScheduler和TaskScheduler。

DAGScheduler把一个spark作业转换成成stage的DAG（Directed Acyclic Graph有向无环图），根据RDD和stage之间的关系，找出开销最小的调度方法，然后把stage以TaskSet的形式提交给TaskScheduler。TaskScheduler模块用于与DAGScheduler交互，负责任务的具体调度和运行。任务调度模块基于两个Trait：TaskScheduler和 SchedulerBackend。

Storage

Storage模块主要分为两层：

• 通信层：Storage模块采用的是master-slave结构来实现通信层，master和slave之间传输控制信息、状态信息，这些都是通过通信层来实现的。
• 存储层：Storage模块需要把数据存储到disk或是memory上面，有可能还需replicate到远端，这都是由存储层来实现和提供相应接口。

Storage模块提供了统一的操作类BlockManager，外部类与storage模块打交道都需要通过调用BlockManager相应接口来实现。Storage模块存取的最小单位是数据块(Block)，Block与RDD中的Partition一一对应，所以所有的转换或动作操作最终都是对Block进行操作。

Shuffle

Shuffle 中Map任务产生的结果会根据所设置的partitioner算法填充到当前执行任务所在机器的每个桶中。Reduce任务启动时时，会根据任务的ID，所依赖的Map任务ID以及MapStatus从远端或本地的BlockManager获取相应的数据作为输入进行处理。Shuffle数据必须持久化磁盘，不能缓存在内存。

Spark之上提供了四种应用库：

• Spark SQL 是为处理结构化数据而设计的模块
• Spark Streaming 可以很容易地创建可扩展和容错性的流式应用
• MLlib 是Spark的可扩展机器学习库
• GraphX 是Spark的并行图计算API

以上便是对Hadoop、Spark的一些浅显的介绍。事实上，解决大数据处理相关的问题，往往要经过数据收集（Flume、Kafka、Sqoop）、数据存储（HDFS、HBase）、资源管理（YARN）、计算（MapReduce、Spark）、数据分析（Pandas、NumPy、R）、数据可视化（Echart、Tableau）等环节。各个环节都有其相应的工具，Hadoop和Spark就是大数据处理流程中非常常用的两个框架。