大数据组件原理

HIVE

Hive是构建在Hadoop上的数据仓库软件框架，支持使用SQL来读，写和管理大规模数据集合。

Hive的服务端组件

1. Driver组件：该组件包括：Compiler、Optimizer、Executor,它可以将Hive的编译、解析、优化转化为MapReduce任务提交给Hadoop1中的JobTracker或者是Hadoop2中的SourceManager来进行实际的执行相应的任务。

2. MetaStore组件：存储着hive的元数据信息，将自己的元数据存储到了关系型数据库当中，支持的数据库主要有：Mysql、Derby、支持把metastore独立出来放在远程的集群上面，使得hive更加健壮。元数据主要包括了表的名称、表的列、分区和属性、表的属性（是不是外部表等等）、表的数据所在的目录。

 3. 用户接口：CLI（Command Line Interface)(常用的接口：命令行模式）、Client:Hive的客户端用户连接至Hive Server ,在启动Client的时候，需要制定Hive Server所在的节点，并且在该节点上启动Hive Server、WUI:通过浏览器的方式访问Hive。

 

Hive的工作原理

如图：

 

流程大致步骤为：

 1. 用户提交查询等任务给Driver。

 2. 编译器获得该用户的任务Plan。

 3. 编译器Compiler根据用户任务去MetaStore中获取需要的Hive的元数据信息。

 4. 编译器Compiler得到元数据信息，对任务进行编译，先将HiveQL转换为抽象语法树，然后将抽象语法树转换成查询块，将查询块转化为逻辑的查询计划，重写逻辑查询计划，将逻辑计划转化为物理的计划（MapReduce）, 最后选择最佳的策略。

 5. 将最终的计划提交给Driver。

 6. Driver将计划Plan转交给ExecutionEngine去执行，获取元数据信息，提交给JobTracker或者SourceManager执行该任务，任务会直接读取HDFS中文件进行相应的操作。

 7. 获取执行的结果。

 8. 取得并返回执行结果。

 Hive中所有的数据都存储在HDFS中，Hive中包含4中数据模型：Tabel、ExternalTable、Partition、Bucket。

 

Impala原理及其调优

Impala执行的查询有以下几个步骤：

1.客户端通过ODBC、JDBC、或者Impala shell向Impala集群中的任意节点发送SQL语句，这个节点的impalad实例作为这个查询的协调器（coordinator）。

2.Impala解析和分析这个查询语句来决定集群中的哪个impalad实例来执行某个任务。

3.HDFS和HBase给本地的impalad实例提供数据访问。

4.各个impalad向协调器impalad返回数据，然后由协调器impalad向client发送结果集。 

 

Impala为什么比Hive速度快

Impala自称数据查询效率比Hive快几倍甚至数十倍，它之所以这么快的原因大致有以下几点：

•真正的MPP查询引擎。

 •使用C++开发而不是Java，降低运行负荷。

•运行时代码生成（LLVM IR），提高效率。

•全新的执行引擎（不是Mapreduce）。

•在执行SQL语句的时候，Impala不会把中间数据写入到磁盘，而是在内存中完成了所有的处理。

•使用Impala的时候，查询任务会马上执行而不是生产Mapreduce任务，这会节约大量的初始化时间。

•Impala查询计划解析器使用更智能的算法在多节点上分布式执行各个查询步骤，同时避免了sorting和shuffle这两个非常耗时的阶段，这两个阶段往往是不需要的。

•Impala拥有HDFS上面各个data block的信息，当它处理查询的时候能够在各个datanode上面更均衡的分发查询。

•另外一个关键原因是，Impala为每个查询产生汇编级的代码，当Impala在本地内存中运行的时候，这些汇编代码执行效率比其它任何代码框架都更快，因为代码框架会增加额外的延迟。 

 

Impala核心组件

Impala Daemon

Impala的核心组件是运行在各个节点上面的impalad这个守护进程（Impala daemon），它负责读写数据文件，接收从impala-shell、Hue、JDBC、ODBC等接口发送的查询语句，并行化查询语句和分发工作任务到Impala集群的各个节点上，同时负责将本地计算好的查询结果发送给协调器节点（coordinator node）。

 

你可以向运行在任意节点的Impala daemon提交查询，这个节点将会作为这个查询的协调器（coordinator node），其他节点将会传输部分结果集给这个协调器节点。由这个协调器节点构建最终的结果集。在做实验或者测试的时候为了方便，我们往往连接到同一个Impala daemon来执行查询，但是在生产环境运行产品级的应用时，我们应该循环（按顺序）的在不同节点上面提交查询，这样才能使得集群的负载达到均衡。

 

Impala daemon不间断的跟statestore进行通信交流，从而确认哪个节点是健康的能接收新的工作任务。它同时接收catalogd daemon（从Impala 1.2之后支持）传来的广播消息来更新元数据信息，当集群中的任意节点create、alter、drop任意对象、或者执行INSERT、LOAD DATA的时候触发广播消息。 

 

 Impala Statestore

Impala Statestore检查集群各个节点上Impala daemon的健康状态，同时不间断地将结果反馈给各个Impala daemon。这个服务的物理进程名称是statestored，在整个集群中我们仅需要一个这样的进程即可。如果某个Impala节点由于硬件错误、软件错误或者其他原因导致离线，statestore就会通知其他的节点，避免其他节点再向这个离线的节点发送请求。

 

由于statestore是当集群节点有问题的时候起通知作用，所以它对Impala集群并不是有关键影响的。如果statestore没有运行或者运行失败，其他节点和分布式任务会照常运行，只是说当节点掉线的时候集群会变得没那么健壮。当statestore恢复正常运行时，它就又开始与其他节点通信并进行监控。

 

Impala Catalog

Impala catalog服务将SQL语句做出的元数据变化通知给集群的各个节点，catalog服务的物理进程名称是catalogd，在整个集群中仅需要一个这样的进程。由于它的请求会跟statestore daemon交互，所以最好让statestored和catalogd这两个进程在同一节点上。

 

Impala 1.2中加入的catalog服务减少了REFRESH和INVALIDATE METADATA语句的使用。在之前的版本中，当在某个节点上执行了CREATE DATABASE、DROP DATABASE、CREATE TABLE、ALTER TABLE、或者DROP TABLE语句之后，需要在其它的各个节点上执行命令INVALIDATE METADATA来确保元数据信息的更新。同样的，当你在某个节点上执行了INSERT语句，在其它节点上执行查询时就得先执行REFRESH table_name这个操作，这样才能识别到新增的数据文件。需要注意的是，通过Impala执行的操作带来的元数据变化，有了catalog就不需要再执行REFRESH和INVALIDATE METADATA，但如果是通过Hive进行的建表、加载数据，则仍然需要执行REFRESH和INVALIDATE METADATA来通知Impala更新元数据信息。

 

 

Hadoop Yarn 框架原理及运作机制

 

1.1 YARN 基本架构

 YARN是Hadoop 2.0中的资源管理系统，它的基本设计思想是将MRv1中的JobTracker拆分成了两个独立的服务：一个全局的资源管理器ResourceManager和每个应用程序特有的ApplicationMaster。

 其中ResourceManager负责整个系统的资源管理和分配，而ApplicationMaster负责单个应用程序的管理。

 

 1.2 YARN基本组成结构

 YARN总体上仍然是Master/Slave结构，在整个资源管理框架中，ResourceManager为Master，NodeManager为Slave，ResourceManager负责对各个NodeManager上的资源进行统一管理和调度。当用户提交一个应用程序时，需要提供一个用以跟踪和管理这个程序的ApplicationMaster，它负责向ResourceManager申请资源，并要求NodeManger启动可以占用一定资源的任务。由于不同的ApplicationMaster被分布到不同的节点上，因此它们之间不会相互影响。在本小节中，我们将对YARN的基本组成结构进行介绍。

 

YARN的基本组成结构，YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等几个组件构成。

 

1.ResourceManager（RM）

 RM是一个全局的资源管理器，负责整个系统的资源管理和分配。它主要由两个组件构成：调度器（Scheduler）和应用程序管理器（Applications Manager，ASM）。

 

（1）调度器

 

调度器根据容量、队列等限制条件（如每个队列分配一定的资源，最多执行一定数量的作业等），将系统中的资源分配给各个正在运行的应用程序。

 

需要注意的是，该调度器是一个“纯调度器”，它不再从事任何与具体应用程序相关的工作，比如不负责监控或者跟踪应用的执行状态等，也不负责重新启动因应用执行失败或者硬件故障而产生的失败任务，这些均交由应用程序相关的ApplicationMaster完成。调度器仅根据各个应用程序的资源需求进行资源分配，而资源分配单位用一个抽象概念“资源容器”（Resource Container，简称Container）表示，Container是一个动态资源分配单位，它将内存、CPU、磁盘、网络等资源封装在一起，从而限定每个任务使用的资源量。此外，该调度器是一个可插拔的组件，用户可根据自己的需要设计新的调度器，YARN提供了多种直接可用的调度器，比如Fair Scheduler和Capacity Scheduler等。

 

（2）应用程序管理器

 应用程序管理器负责管理整个系统中所有应用程序，包括应用程序提交、与调度器协商资源以启动ApplicationMaster、监控ApplicationMaster运行状态并在失败时重新启动它等。

 

2. ApplicationMaster（AM）

 用户提交的每个应用程序均包含1个AM，主要功能包括：

 与RM调度器协商以获取资源（用Container表示）；

 将得到的任务进一步分配给内部的任务；

 与NM通信以启动/停止任务；

 监控所有任务运行状态，并在任务运行失败时重新为任务申请资源以重启任务。

 当前YARN自带了两个AM实现，一个是用于演示AM编写方法的实例程序distributedshell，它可以申请一定数目的Container以并行运行一个Shell命令或者Shell脚本；另一个是运行MapReduce应用程序的AM—MRAppMaster，此外，一些其他的计算框架对应的AM正在开发中，比如Open MPI、Spark等。

 

3. NodeManager（NM）
NM是每个节点上的资源和任务管理器，一方面，它会定时地向RM汇报本节点上的资源使用情况和各个Container的运行状态；另一方面，它接收并处理来自AM的Container启动/停止等各种请求。

 

4. Container

 

Container是YARN中的资源抽象，它封装了某个节点上的多维度资源，如内存、CPU、磁盘、网络等，当AM向RM申请资源时，RM为AM返回的资源便是用Container表示的。YARN会为每个任务分配一个Container，且该任务只能使用该Container中描述的资源。

 

需要注意的是，Container不同于MRv1中的slot，它是一个动态资源划分单位，是根据应用程序的需求动态生成的。截至本书完成时，YARN仅支持CPU和内存两种资源，且使用了轻量级资源隔离机制Cgroups进行资源隔离。

 

 

 

1.3  YARN工作流程

 

当用户向YARN中提交一个应用程序后，YARN将分两个阶段运行该应用程序：

 第一个阶段是启动ApplicationMaster；

 第二个阶段是由ApplicationMaster创建应用程序，为它申请资源，并监控它的整个运行过程，直到运行完成。

 YARN的工作流程分为以下几个步骤：

 步骤1　用户向YARN中提交应用程序，其中包括ApplicationMaster程序、启动ApplicationMaster的命令、用户程序等。

 

步骤2　ResourceManager为该应用程序分配第一个Container，并与对应的Node-Manager通信，要求它在这个Container中启动应用程序的ApplicationMaster。

 

步骤3　ApplicationMaster首先向ResourceManager注册，这样用户可以直接通过ResourceManager查看应用程序的运行状态，然后它将为各个任务申请资源，并监控它的运行状态，直到运行结束，即重复步骤4~7。

 

步骤4　ApplicationMaster采用轮询的方式通过RPC协议向ResourceManager申请和领取资源。

 

步骤5　一旦ApplicationMaster申请到资源后，便与对应的NodeManager通信，要求它启动任务。

 

步骤6　NodeManager为任务设置好运行环境（包括环境变量、JAR包、二进制程序等）后，将任务启动命令写到一个脚本中，并通过运行该脚本启动任务。

 

步骤7　各个任务通过某个RPC协议向ApplicationMaster汇报自己的状态和进度，以让ApplicationMaster随时掌握各个任务的运行状态，从而可以在任务失败时重新启动任务。

在应用程序运行过程中，用户可随时通过RPC向ApplicationMaster查询应用程序的当前运行状态。

 

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

 

 1.4 多角度理解YARN

 

可将YARN看做一个云操作系统，它负责为应用程序启动ApplicationMaster（相当于主线程），然后再由ApplicationMaster负责数据切分、任务分配、启动和监控等工作，而由ApplicationMaster启动的各个Task（相当于子线程）仅负责自己的计算任务。当所有任务计算完成后，ApplicationMaster认为应用程序运行完成，然后退出。

 Spark on Yarn的运行原理

 

一、YARN是集群的资源管理系统

 1、ResourceManager：负责整个集群的资源管理和分配。

 2、ApplicationMaster：YARN中每个Application对应一个AM进程，负责与RM协商获取资源，获取资源后告诉NodeManager为其分配并启动Container。

 3、NodeManager：每个节点的资源和任务管理器，负责启动/停止Container，并监视资源使用情况。

 4、Container：YARN中的抽象资源。

 

二、SPARK的概念

 1、Driver：和ClusterManager通信，进行资源申请、任务分配并监督其运行状况等。

 2、ClusterManager：这里指YARN。

 3、DAGScheduler：把spark作业转换成Stage的DAG图。

 4、TaskScheduler：把Task分配给具体的Executor。

 

三、SPARK on YARN

 3.1、yarn-cluster模式下

(1)ResourceManager接到请求后在集群中选择一个NodeManager分配Container，并在Container中启动ApplicationMaster进程

 (2)在ApplicationMaster进程中初始化sparkContext；

 (3)ApplicationMaster向ResourceManager申请到Container后，通知NodeManager在获得的Container中启动excutor进程；

 (4)sparkContext分配Task给excutor，excutor发送运行状态给ApplicationMaster。

 

3.2、yarn-clinet模式下

(1)ResourceManager接到请求后在集群中选择一个NodeManager分配Container，并在Container中启动ApplicationMaster进程；

 

(2)driver进程运行在client中，并初始化sparkContext；

 

(3)sparkContext初始化完后与ApplicationMaster通讯，通过ApplicationMaster向ResourceManager申请Container，ApplicationMaster通知NodeManager在获得的Container中启动excutor进程；

 

(4)sparkContext分配Task给excutor，excutor发送运行状态给driver。

 

3.3、yarn-cluster与yarn-client的区别：

 它们的区别就是ApplicationMaster的区别，yarn-cluster中ApplicationMaster不仅负责申请资源，并负责监控Task的运行状况，因此可以关掉client；

 而yarn-client中ApplicationMaster仅负责申请资源，由client中的driver来监控调度Task的运行，因此不能关掉client。

  

spark RDD底层原理

 

RDD底层实现原理

 RDD是一个分布式数据集，顾名思义，其数据应该分部存储于多台机器上。事实上，每个RDD的数据都以Block的形式存储于多台机器上，下图是Spark的RDD存储架构图，其中每个Executor会启动一个BlockManagerSlave，并管理一部分Block；而Block的元数据由Driver节点的BlockManagerMaster保存。BlockManagerSlave生成Block后向BlockManagerMaster注册该Block，BlockManagerMaster管理RDD与Block的关系，当RDD不再需要存储的时候，将向BlockManagerSlave发送指令删除相应的Block。

 

图2 RDD存储原理

 

RDD cache的原理

 RDD的转换过程中，并不是每个RDD都会存储，如果某个RDD会被重复使用，或者计算其代价很高，那么可以通过显示调用RDD提供的cache()方法，把该RDD存储下来。那RDD的cache是如何实现的呢？

 

RDD中提供的cache()方法只是简单的把该RDD放到cache列表中。当RDD的iterator被调用时，通过CacheManager把RDD计算出来，并存储到BlockManager中，下次获取该RDD的数据时便可直接通过CacheManager从BlockManager读出。

 

RDD dependency与DAG

     RDD提供了许多转换操作，每个转换操作都会生成新的RDD，这是新的RDD便依赖于原有的RDD，这种RDD之间的依赖关系最终形成了DAG（Directed Acyclic Graph）。

 

    RDD之间的依赖关系分为两种，分别是NarrowDependency与ShuffleDependency，其中ShuffleDependency为子RDD的每个Partition都依赖于父RDD的所有Partition，而NarrowDependency则只依赖一个或部分的Partition。下图的groupBy与join操作是ShuffleDependency，map和union是NarrowDependency。

  

图3 RDD dependency

 RDD partitioner与并行度

 

    每个RDD都有Partitioner属性，它决定了该RDD如何分区，当然Partition的个数还将决定每个Stage的Task个数。当前Spark需要应用设置Stage的并行Task个数（配置项为：spark.default.parallelism），在未设置的情况下，子RDD会根据父RDD的Partition决定，如map操作下子RDD的Partition与父Partition完全一致，Union操作时子RDD的Partition个数为父Partition个数之和。

 

    如何设置spark.default.parallelism对用户是一个挑战，它会很大程度上决定Spark程序的性能。