YARN知识点总结

Posted 2023-03-21

参考技术A 如果把大数据Hadoop集群当作一台计算机, 那么

HDFS = 磁盘

YARN = 任务调度器+资源管理器

所有任务都是运行在Yarn上

Yarn分为两个大的模块: ResourceManager NodeManager

运行在master机器上, 用于分配资源

两个模块

Scheduler  负责资源分配

ApplicationsManager 负责应用管理.

RM不负责启动container, 而是告诉AM哪些container可用, 由AM向NM发起请求来启动container.

RM负责启动AM, 并且在应用程序失败时, 重启AM.

运行在slave机器上, 用于执行任务

container是计算资源, 代表任务执行时所分配的cpu,内存等资源.

每个任务都是在container里执行的.

执行过程

client首先向RM提交应用, 请求一个AM实例.

RM找到一个可以启动container的NM, 并启动一个AM实例

AM向RM注册之后, client就可以从RM查找到AM的详细信息, 可以和AM建立联系.

AM向RM的Scheduler发送resource-request请求, Scheduler返回资源描述, 包括资源名称(NM的地址,rack等)、优先级、资源需求和container集合

AM向对应的NM发起container-launch-specification来启动container

应用程序在container中运行, 并把进度、状态等信息通过application-specific协议发送给AM

应用程序执行完毕, AM向RM注销, 并回收对应的container

scheduler是个可插拔的组件, Yarn常用3种scheduler

FIFO Scheduler

完全先进先出, 按照提交的先后顺序. 大任务会堵塞小任务. 不适合共享集群

Capacity Scheduler

为每个团队分配单独的队列, 每个队列分配对应的资源. 每个队列内部使用FIFO方式.

如果资源不够, 可以使用其他队列释放出来的资源, 但是需要设置队列能够占用的最大资源, 防止队列资源占用过高.

Fair Scheduler

根据权重来分配各队列可以使用的资源.

如果一个队列没有任务, 那么这个队列的资源可以被其他队列使用.

否则按照设置好的权重来分配队列的可用资源.

没有设置权重的队列则使用平均分配的方式分配资源

抢占preemption

为了执行任务, 把超出自己份额的应用的container杀掉, 用来执行新提交的任务.

Node Status: Node Manager和Resource Manager之间保持心跳, 并发送资源使用情况给RM.

Hadoop（HDFSYarnMapReduceZookeeperKafkaFlumeStormSpark） 知识点总结

知识点总结

• • 请简述HDFS集群的总体架构以及NameNode、DataNode和SecondaryNameNode的作用。
  • 什么是HDFS的机架感知策略？在HDFS使用该策略有什么优点？
  • 简述HDFS读写数据的流程。
  • 简述YARN集群的总体架构以及ResourceManager、NodeManager、ApplicationMaster和Container的作用。
  • 以单词计数为例，简述MapReduce计算模型的三个阶段。
  • 在ZooKeeper集群中，znode节点有哪几种类型？不同类型的节点分别具有什么特点？
  • 什么是HDFS单点故障问题？HDFS HA是如何解决HDFS单点故障问题的？
  • 简述Kafka集群架构。
  • 请简述Kafka主题与分区的概念。
  • 请简述Kafka分区副本同步的基于ISR动态复制方案及其优点。
  • 请简述Kafka消费者组的含义及其作用。
  • 举例说明自定义Kafka生产者、消费者？
  • Kafka生产者拦截器有什么作用？如何常见生产者拦截器
  • 简述Flume Agent的主要构建组件及其作用。
  • Flume拦截器有什么作用？
  • 简述Flume通道选择器的作用及主要类型。
  • 请简述Storm Topology的基本概念及其构成。
  • 请简述Storm的集群架构。
  • 请简述Storm流分组的概念及其常见方式。
  • 请简述Spark的主要组件及其主要功能。
  • Spark的部署模式有哪几种？
  • 什么是RDD？创建RDD主要由哪几种方式？请举例说明。
  • 请简述RDD的转化算子和行动算子的主要区别？
  • 什么是Spark惰性机制?这种机制有何优缺点？
  • 请列出RDD的主要算子及其作用。
  • 什么是DataFrame？DataFrame与RDD有什么不同？
  • 创建DataFrame主要由哪几种方式？请举例说明。
  • 请举例说明如何在Spark中使用SQL进行查询。
  • 请简述Spark Streaming的工作机制。
  • 请比较Spark Streaming和Storm有何不同？
  • 请简述编写Spark Streaming程序的主要步骤。
  • 请请举例说明DStream主要的有状态转换操作及其作用。

请简述HDFS集群的总体架构以及NameNode、DataNode和SecondaryNameNode的作用。

​ HDFS采用Master/Slave（主/从）架构：即一个HDFS集群是由一个NameNode和若干个DataNode组成的。

​ NameNode是存储集群的主服务器，负责管理文件系统的命名空间（NameSpace）以及客户端对文件的访问

​ DataNode负责处理文件系统客户端的读写。在NameNode的统一调度下进行数据块的创建、删除和复制操作。

​ HDFS的辅助元数据节点(SecondaryNameNode)辅助NameNode处理事务日志和镜像文件。

什么是HDFS的机架感知策略？在HDFS使用该策略有什么优点？

​ 以默认的副本数=3为例

​ 第一个副本块存本地机架的节点上

​ 第二个副本块存跟本机同机架内的其他节点上

​ 第三个副本块存不同机架的节点上

​ 优点：

​ 减少了机架间的数据传输，提高了效率

​ 不损坏数据的可靠性和读取性能 机架错误远远比节点错误要小

简述HDFS读写数据的流程。

​ 读文件：客户端要读某个文件

1. 客户端向NameNode发送数据读操作请求
2. NameNode向客户端发送组成该文件的数据块的位置列表（即每个数据块存储哪些DataNode之中）
3. 客户端直接从这些DataNode读取文件数据（在读数据过程中，NameNode不参与文件的传输）

​

​ 写文件：当客户端需要写入一个文件

1. 客户端向NameNode发送数据写操作请求，将需要写入的文件名、路径等元数据信息（MetaData）告诉NameNode
2. NameNode将文件信息记录到本地，同时验证客户端的写入权限，若验证通过，会向客户端返回文件数据块能够存放在DataNode上的存储位置信息。
3. 客户端直接向DataNode的相应位置写入数据块。
4. 被写入的数据块的DataNode也会将数据块备份到其他DataNode上。

简述YARN集群的总体架构以及ResourceManager、NodeManager、ApplicationMaster和Container的作用。

​ YARN集群采用经典的Master/Slave（主/从）架构

​ RescourseManager负责对集群资源的统一管理和任务调度

​ NodeManager是集群中每个节点上的资源和任务管理器

​ ApplicationMaster是应用程序管理者，主要负责应用程序的管理

​ Container是YARN中资源分配的基本单位，封装了CPU和内存资源的一个容器，相当于一个Task运行的抽象环境。

以单词计数为例，简述MapReduce计算模型的三个阶段。

​ 首先在map阶段，需要将接收到的数据来进行拆分，接着将每个单词统计一次数量；

​ 然后到shuffle阶段处理，将不同单词中一样的单词分到同一组中；

​ 最后是reduce阶段，将每一组单词中的数量添加到一起，最后再整合输出。

在ZooKeeper集群中，znode节点有哪几种类型？不同类型的节点分别具有什么特点？

​ Znode有四种类型：分别是持久节点（PERSISTENT）、持久顺序节点（PERSISTENT_SEQUENTIAL）、临时节点（EPHEMERAL）以及临时顺序节点（EPHEMERAL_SEQUENTIAL）

​ 持久节点（PERSISTENT）在创建后会一直存在，除非手动将其删除

​ 持久顺序节点（PERSISTENT_SEQUENTIAL）在拥有持久节点的功能的同时，在创建时，ZooKeeper会在节点名称末尾自动追加一个自增长的数字后缀作为新的节点名称，以便记录每个节点创建的先后顺序。

​ 临时节点（EPHEMERAL）客户端与ZooKeeper服务器会话存在，这些节点就存在，客户端会话结束时，节点将被删除；以及临时节点不可以有子节点。

什么是HDFS单点故障问题？HDFS HA是如何解决HDFS单点故障问题的？

​ 一个HDFS集群只有一个单一的NameNode，一旦NameNode服务不可用，则整个集群无法访问

​ 可以在同一个集群中运行两个NameNode，其中一个处于活动状态（active），另一个处于备用状态(standby),且只有活动状态的NameNode可以对外提供读写服务。当活动状态的NameNode崩溃时，HDFS集群可以快速切换到备用的NameNode，这样也就是实现了故障自动转移

简述Kafka集群架构。

Kafka集群架构：

• 一组生产者
  • 数据可以是Web前端产生的页面、服务器日志等
• 一组消费者
  • 可以是Hadoop集群、实时监控程序、数据仓库或其他服务
• 一组Broker
• 一个ZooKeeper集群

请简述Kafka主题与分区的概念。

主题：

• 一个主题可以分为多个分区，每个分区可以存储于不同的Broker上

• Kafka会为每个主题维护一个分区日志，记录各个分区消息存放情况

• 消息以追加的方式写入到每个分区的尾部

• 消息以先入先出的顺序进行读取

• Kafka可以保证单个分区内消息的顺序，但无法在整个主题范围内保证消息的顺序

分区：

• 每条消息会根据分区规则被存储到某个分区

• 如果分区合理，所有消息可以被均匀分配到不同的分区，这样就实现了水平扩展

• 分区中的每条记录被都分配了一个偏移量（offset）

• 偏移量是一个连续递增的整数，唯一标识分区中的某个记录

• 消费者只需要保存偏移量，当消费者客户端向Broker发起消息请求时需要携带偏移量，当消费者读取消息后，偏移量会递增

• 消费者可以按照任意顺序消费消息，也可以指定从某个分区中一次最多返回多少条消息

请简述Kafka分区副本同步的基于ISR动态复制方案及其优点。

基于ISR(In-sync Replica)动态复制方案

领导者会动态维护一个需要与其保持同步的副本列表（包括领导者自己），该列表称为ISR。

如果在一定时间内跟随者没有向领导者请求新的消息，该跟随者将被认为不同步，领导者会从ISR中将其移除（防止拖慢整体速度）

当跟随者重新与领导者保持同步时，领导者会将其再次加入到ISR中

当领导者失效时，也不会选择ISR中不存在的跟随者作为新的领导者

ISR的列表数据保存在ZooKeeper中，每次ISR改变后，领导者都会将最新的ISR同步到ZooKeeper中。

优点：

写入消息时，只有当ISR中所有跟随者都复制完毕，领导者才会将信息写入状态置为Commit（写入成功），而只有状态置为Commit的消息才能被消费者读取

从消费者角度看，要想成功读取消息，ISR中的所有副本必须处于同步状态，从而提高了数据的一致性

请简述Kafka消费者组的含义及其作用。

含义：

消费者组（Consumer Group）实际上就是一组消费者集合

作用：

能够同时具备两种模式（队列和发布订阅）的特点

同一消费者组内不允许多个消费者消费同一分区的消息，不同的消费者组可以同时消费同一分区的消息

分区与同一消费者组中的消费者是多对一的关系

举例说明自定义Kafka生产者、消费者？

自定义生产者消费者，可以根据开发者的需求，定制特殊的生产者生产有用的信息，定制特殊的消费者来接受特定主题的信息，然后再进行进一步的处理操作。

Kafka生产者拦截器有什么作用？如何常见生产者拦截器

生产者拦截器：

在消息发送前对消息内容进行定制化修改，以便满足相应的业务需求，也可以用于在消息发送后获取消息的发送状态、所在分区和偏移量等信息

用户可以在生产者中指定多个拦截器形成一个拦截器链，生产者会根据指定顺序先后调用

需要实现生产者接口

常见生产者拦截器：

• 时间戳拦截器
• 统计拦截器

简述Flume Agent的主要构建组件及其作用。

Flume Agent三大组件

• Source（源）

  • 收集数据，并传递给Channel

• Channel（通道）

  • 将Source传输的数据暂时存放

• Sink（接收地）

  • 从Channel接收数据，并写入到指定地址

Flume拦截器有什么作用？

拦截器（Interceptor）

作用：修改或删除正在传送中event

拦截器是一些实现Interceptor接口的类

在Source组件中设置，支持设置多个拦截器

多个拦截器使用空格连接在一起，根据配置顺序依次执行

如果某个拦截器需要删除event，当event经过该拦截器后，该event会被过滤掉，不会返回给下一个拦截器

简述Flume通道选择器的作用及主要类型。

Source可以将event写入多个Channel，而Channel选择器可以决定将event写入哪些Channel

Flume内置两种选择器

复制选择器：将同一个event发送到每个Channel。

多路选择器：按照event的头部配置将event发送到相应的Channel。

请简述Storm Topology的基本概念及其构成。

拓扑（Topology）

一个实时计算任务被称为拓扑（Topology）

拓扑（Topology）的主要构成组件

Spout：数据源

Bolt：数据处理

Tuple：消息传递的一个基本单元

Stream：源源不断的Tuple组成了Stream

请简述Storm的集群架构。

分布式集群采用经典的主从架构

Nimbus:主节点，运行一个名为Nimbus的主控进程

Supervisor:工作节点，运行一个名为Supervisor的工作进程

客户端提交Topolopy给Nimbus

Nimbus负责分发Topolopy给Supervisor

Nimbus通过Zookeeper监控Supervisor的状态和确定任务分配策略

Supervisor定时与Zookeeper同步

以便获取Topolopy信息、任务分配信息及各类心跳信息

Supervisor根据需要启动一个或多个Worker进程执行具体的Topolopy

每个Worker只能执行一个Topolopy

但同一个Topology可以由多个Worker共同执行

一个Topolopy通常由多个节点的多个Worker共同完成

Supervisor会根据新任务分配情况来调整Worker的数量并进行负载均衡

请简述Storm流分组的概念及其常见方式。

流分组（ Stream grouping ）用于在定义一个Topolopy时，Bolt指定它应该接收哪些Stream作为输入

一个Stream grouping定义了如何在Bolt的多个Task之间划分该Stream ，即对Stream中的Tuple进行分组，使不同Tuple进入不同的Task

常见方式：

Shuffle grouping

Fields grouping

Partial Key grouping

All grouping

Global grouping

None grouping

Direct grouping

Local or shuffle grouping

也可以通过实现CustomStreamGrouping接口来实现自义定流分组

请简述Spark的主要组件及其主要功能。

• **Spark Core：**该组件是Spark的核心模块，主要包含两个功能：一是负责任务调度、内存管理、错误恢复与存储系统交互等；二是其包含了对弹性分布式数据集的API定义。它提供了创建和操作这些集合的多个API。
• **Spark SQL：**该组件是一个用于结构化数据处理的Spark工具包，提供了面向结构化数据的SQL查询接口，使用户可以通过编写SQL或基于Apache Hive的HiveQL来方便地处理数据。也可以查询Hive中的数据，相当于数据仓库的查询引擎，提供了很强大的计算速度。它还支持开发者将SQL语句融入到Spark应用程序的开发过程中，使得用户可以用SQL语句进行查询和复杂的数据分析。
• **Spark Steaming：**该组件是Spark提供的对实时数据进行流式计算的组件，它将流式的计算分解成一系列短小的批处理作业，支持对实时数据流进行可伸缩、高吞吐量、容错的流处理。它还提供了用来操作数据流的API，并且与Spark Core当中的RDD API高度对应，帮助开发人员高效地处理数据流中的数据。且支持与Spark Core同级别的容错性、吞吐量及可伸缩性。Spark Streaming通过将数据流中的数据按指定时间片累计为RDD，然后将每个RDD进行批处理，进而实现大规模的流数据处理。
• **MLlib：**该组件是Spark的机器学习库。它的目标是使得机器学习具有可拓展性和易用性。其中提供了分类、回归、聚类、协同过滤等常用机器学习算法，以及一系列更加底层的机器学习原语。
• **GraphX：**该组件是Spark中图形和图形并行计算的一个新㢟，可以用其创建一个顶点和边都包含任意属性的有向多重图。此外，该组件还包含越来越多的图算法和构建起，以此来简化图形分析任务。

Spark的部署模式有哪几种？

三种部署模式：本地（单机）模式、Yarn集群模式、Spark自带的Standalone模式。

​ **本地模式：**是只运行在一台计算机上，就称为本地模式。

​ **Yarn集群模式：**Spark On Yarn模式遵循YARN的官方规范，YARN只负责资源的管理和调度，运行哪种应用程序由用户自己实现，因此可能在YARN上同时运行MapReduce程序和Spark程序，YARN很好地对每一个程序实现了资源的隔离。这使得Spark与MapReduce可以运行于同一个集群中，共享集群存储资源与计算资源。

​ **Standalone模式：**该模式采用经典的Master/Slave架构，资源调度由Spark自己实现。

什么是RDD？创建RDD主要由哪几种方式？请举例说明。

​ RDD是Spark提供的一种对数据的核心抽象，称为弹性分布式数据集（Resilient Distributed Dataset，RDD）。每个RDD被分为多个分区，这些分区运行在集群中的不同节点上。也就是说，RDD是跨集群节点分区的元素集合，并且这些元素可以并行操作。

​ 创建RDD主要有两种方式，一是从文件系统中加载数据创建RDD，二是通过对象集合（数组）创建RDD。

1. 从对象集合创建RDD：将一个List集合转化为RDD：

val rdd = sc.parallelize(List(1,2,3,4,5))

​ 2.从文件系统中加载数据创建RDD：

val rdd = sc.textFile(“file:///home/words.txt”)

请简述RDD的转化算子和行动算子的主要区别？

**转化算子：**Spark中转化算子不会立即进行运算，它是惰性的，只是记住对某个RDD的具体操作过程。

**行动算子：**Spark中遇到行动算子则会执行相应的语句，触发任务调度。

什么是Spark惰性机制?这种机制有何优缺点？

​ Spark惰性机制指整个转化过程只是记录了转化的轨迹，并不会发生真正的计算，只有遇到行动操作时，才会发生真正的计算，开始从血缘关系源头开始，进行物理的转化操作。这样可以避免许多无意义的计算，节省内存。但是每次遇到行动操作，都会从头开始执行计算。每次调用行动操作，都会触发一次从头开始的计算。这对于迭代计算而言，代价是很大的，迭代计算经常需要多次重复使用同一组数据。

请列出RDD的主要算子及其作用。

转化算子:

• **Map()**算子：接收一个函数作为参数，并把该函数作用于RDD的每个元素，最后将函数的返回结果作为结果RDD中对应元素的值。

• **Filter(func)**算子：通过函数func对源RDD的每个元素进行过滤，并返回一个新的RDD。

• **FlatMap(func)**算子：与map()算子类似，但每个传入给函数func的RDD元素会返回0到多个元素，最终会将返回的所有元素合并到一个RDD中。

• **ReduceByKey(func)**算子：作用对象元素为（key，value）形式的RDD，可以将相同的key的元素聚集到一起，最终把所有Key相同的元素合并成一个元素。该元素的key值不变，value可以聚合成一个列表或者进行求和等操作。最终返回的RDD的元素类型和原有类型保持一致。

• **Union()**算子：该算子能将两个RDD合并成为一个新的RDD，主要用于对不同的数据来源进行合并，两个RDD中的数据类型要保持一致。

• **SortBy(func)**算子：该算子可以将RDD中的元素按照某个规则进行排序。

行动算子

• **Reduce(func)**算子：将RDD中的元素进行聚合计算。

• **Count()**算子：返回数据集中元素的数量。

• **CountByKey()**算子：统计RDD中key相同的元素的数量。

• **Take(n)**算子：返回包含数据集的前n个元素的数组。

什么是DataFrame？DataFrame与RDD有什么不同？

​ DataFrame是Spark SQL提供的一个编程抽象，与RDD类似，也是一个分布式的数据集合。但不同的是，DataFrame的数据都被组织到有名字的列中，在RDD的基础上添加了数据描述信息，就像关系型数据库中的表一样。此外，多种数据都可以转化为DataFrame。

创建DataFrame主要由哪几种方式？请举例说明。

1、用SparkSession从不同的数据源中加载数据，并转化成DataFrame。

举例：

val peopleDF = spark.read.format("json").load("file:///spark/examples/people.json")

peopleDF.select("name", "age").write.format("csv").save("file:///usr/local/newpeople.csv")

2、从RDD转化得到DataFrame

举例：

//生成字段
val fields = Array(StructField("name",StringType,true),StructField("age",IntegerType,true))

val schema = StructType(fields)

val peopleRDD = spark.sparkContext.textFile("file:///spark/examples/people.txt")

val rowRDD = peopleRDD.map(_.split(",")).map(attributes => Row(attributes(0), attributes(1).trim.toInt))

val peopleDF = spark.createDataFrame(rowRDD, schema)

反射机制推断RDD模式：

case class Person(name:String,age:Long) //定义一个case class 
val peopleDF = spark.sparkContext.textFile("file:///spark/examples/people.txt").map(_.split(",")).
map(attributes=>Person(attributes(0),attributes(1).trim.toInt)).toDF()

//注册临时表
peopleDF.createOrReplaceTempView("people")

请举例说明如何在Spark中使用SQL进行查询。

1. 创建RDD

2. 将RDD转换为DataFrame

3. 创建临时视图

4. 使用sql语句查询

//将RDD转换为DataFrame
val peopleDF = spark.createDataFrame(rowRDD, schema)
//创建临时视图people
peopleDF.createOrReplaceTempView("people")
//调用SQL语句，进行SQL查询
val results = spark.sql("SELECT name,age FROM people")
//打印查询结果
val results.map(attributes => "name: "+ attributes(0)+","+"age:"+attributes(1)).show()

请简述Spark Streaming的工作机制。

​ 在Spark Streaming中，会有一个组件Receiver，作为一个长期运行的task跑在一个Executor上。

​ 每个Receiver都会负责一个input DStream（比如从文件中读取数据的文件流，比如套接字流，或者从Kafka中读取的一个输入流等）。

​ Spark Streaming通过input DStream与外部数据源进行连接，读取相关数据。

请比较Spark Streaming和Storm有何不同？

​ Spark Streaming和Storm最大的区别在于，Spark Streaming无法实现毫秒级的流计算，而Storm可以实现毫秒级响应

​ Spark Streaming构建在Spark上，一方面是因为Spark的低延迟执行引擎（100ms+）可以用于实时计算，另一方面，相比于Storm，RDD数据集更容易做高效的容错处理

​ Spark Streaming采用的小批量处理的方式使得它可以同时兼容批量和实时数据处理的逻辑和算法，因此，方便了一些需要历史数据和实时数据联合分析的特定应用场合

请简述编写Spark Streaming程序的主要步骤。

1. 通过创建输入DStream来定义输入源

2. 通过对DStream应用转化操作和输出操作来定义流计算

3. 用streamingContext.start()来开始接收数据和处理流程

4. 通过streamingContext.awaitTermination()方法来等待处理结束（手动结束或因为错误而结束）

5. 可以通过streamingContext.stop()来手动结束流计算进程

请请举例说明DStream主要的有状态转换操作及其作用。

1. 滑动窗口转化操作

​ 设定一个滑动窗口长度（也就是持续时间），设定窗口滑动时间（也就是间隔多少时间进行一次计算），让窗口按照指定时间在源DStream上滑动。每次窗口停止的位置上，都会有一部分DStream被新增入窗口内，形成一个小段的DStream，可以启动对该段DStream的计算。

2. UpdateStateByKey操作

​ 需要在跨批次之间维护状态时，就必须使用updateStateByKey操作。