大数据常见面试题

1、MySQL索引有哪些？

1.1、普通索引index：加速查找

1.2、唯一索引

主键索引：primary key：加速查找+约束（不为空且唯一）
唯一索引：unique：加速查找+约束（唯一）

1.3、联合索引

primary key(id,name)：联合主键索引
unique(id,name)：联合唯一索引
index(id,name)：联合普通索引

1.4、全文索引fulltext：用于搜索很长一篇文章的时候，效果最好

1.5、空间索引spatial：了解，几乎不用

2、HDFS读写流程？（重点）

2.1、写数据流程

2.1.1、客户端发出请求 hdfs dfs -put /etc/profile/qf/data

2.1.2、namenode查看维护的目录结构，检查/qf/data是否存在，如不存在直接报错”no such file or directory“，如果存在返回给客户端同意上传文件请求，将操作写入日志文件

2.1.3、客户端请求上传第一个块，询问namenode块的存储位置

2.1.4、namenode查看自己的datanode池，返回给客户端一个datanode列表

2.1.5、客户端发出请求建立pipeline

2.1.6、客户端先把文件写入缓存，达到一个块的大小时，会在客户端和第一个datanode建立连接并开始流式的传输数据，这个datanode会一小部分一小部分（4K）的接收数据然后写入本地仓库，同时会将这些数据传输到第二个datanode，第二个datanode也同样一小部分一小部分的接收数据并写入本地仓库，同时传输给第三个datanode…（在流式复制时，逐级传输和响应采用响应队列来等待传输结果。队列响应完成后返回给客户端）

2.1.7、第一个数据块传输完成后会使用同样的方式传输下面的数据块直接到整个文件上传完成。

2.1.8、整个文件完成后，namenode更新内存元数据。

2.2、读数据流程

2.2.1、客户端向namenode发起RPC调用，请求读取文件数据。

2.2.2、namenode检查文件是否存在，如果存在则获取文件的元信息（blockid以及对应的datanode列表）。

2.2.3、客户端收到元信息后选取一个网络距离最近的datanode，依次请求读取每个数据块。客户端首先要校验文件是否损坏，如果损坏，客户端会选取另外的datanode请求。

2.2.4、datanode与客户端建立socket连接，传输对应的数据块，客户端收到数据缓存到本地，之后写入文件。

2.2.5、依次传输剩下的数据块，直到整个文件合并完成。

3、Spark的持久化机制和检查点，区别是什么，应用场景？

3.1、持久化（重点）

spark通过catch和persist方法对结果进行一个持久化，persist方法共有5个参数，对应12个缓存级别，这12个级别分别从磁盘存储、内存存储、堆外内存存储、是否反序列化和备份数五个角度设定。其中catch使用的是Memory_Only，只在内存持久化。

3.2、检查点（重点）

spark通过checkPoint方法将RDD状态保存在高可用存储中，与持久化不同的是，它是对RDD状态的一个复制持久化，执行checkPoint后不再保存依赖链。此外，持久化存储的缓存当程序运行结束后就会被自动删除，检查点保存的RDD状态只能手动清理。

4、Yarn的提交工作原理？

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-rFrR9KfX-1664237369872)(C:\Users\25851\AppData\Roaming\Typora\typora-user-images\image-20220926091207587.png)]

5、Hive的内部表和外部表的区别？（重点）

hive外部表是使用external关键字并指定一个hdfs目录创建的表。

hive内部表在创建时会在对应hive目录下创建相应的文件夹，外部表则以指定文件夹为数据源创建表。

hive内部表在删除时会将整个文件夹一并删除，外部表则只会删除元数据。

6、Hive的OrderBy和SortBy区别？

6.1、order by

需要加载所有数据到reduce中排序，排序方法可能是冒泡、快排、归并，无论如何都要加载所有数据

6.2、sort by

局部排序，最终生成的每个文件都有序但不能保证全局有序

7、Flume会不会发生数据重复，解释说明？

会，我们在获取Channel批次数据的时候，每个批次是基于一个事务性的，一旦发生宕机或者Flume进行挂掉，那么就会回滚，数据写入失败，下次再重启写入时候数据就重复

8、Flume拦截器分为那些？

Source 将 Event 写入到 Channel 之前可以使用拦截器对 Event 进行各种形式的处理，Source 和Channel 之间可以有多个拦截器，不同拦截器使用不同的规则处理 Event，包括时间、主机、UUID、正则表达式等多种形式的拦截器。

9、Sqoop增量导入和全量导入的区别？***

全量全部数据覆盖导入，增量是按照条件进行append追加，一般使用job方式记录last-value，全量一般是做少量更新数据的表，增量一般是做大数据量更新的表

10、数仓与数据库的区别？

操作上：数据库是面向事务的，往往是行级操作;数仓则是面向分析的，往往是范围操作

功能上：数据库提供即时的增删改查，数仓则寻求的是分析数据提供决策支持

设计上：数据库基于ER模型，面向业务;数仓基于星型/雪花模型，面向主题

数据上：数据库只保存最新的、近期的数据;数仓则保存所有数据

性能上：数据库往往依靠索引快速返回;数仓则往往需要大范围磁盘扫描

数据量：数据库数据一般为GB级别，数仓的数据则往往上百TB

响应速度：数据库是毫秒级，数仓任务的执行时间往往数小时

存储上：数据库是真实的物理存储，数仓则是逻辑存储。

11、Spark的累加器是什么？如何应用？***

spark提供了一个累加器用于在整个流程中额外执行一个MR任务，它可以在driver端被初始化发送给各个Task，然后在每个Task中为它添加数据，最终经过reduce将结果聚合后返回driver端。可以自定义累加器的类型，通过实现一个聚合方法来创建自定义累加器。除此之外spark2还支持特殊的累加器-收集器，它不需要执行reduce，会将数据原原本本存放在集合中返回。注意：如果累加操作在transform算子并且action算子有多个时，需要catch该转换算子，否则可能造成重复累加。

12、Redis的持久化策略？

12.1、RDB持久化

符合一定条件时redis会将内存中所有数据以二进制方式生成一份副本存储到硬盘上（快照），redis重启时可以通过该副本恢复数据，生成快照时redis进入阻塞状态。

12.2、AOF持久化

redis将日志追加到磁盘文件中，redis启动时优先选择从AOF文件恢复数据。由于每次写操作都会记录日志，因此AOF会降低性能，但比起RDB的一次性阻塞式备份数据，AOF消耗的内存和资源更少。 AOF文件达到阀值时（或执行bgrewriteaof命令），会重写一次AOF文件，将每个Key只保留最新的value的日志

12.3、混合持久化

开启混合持久化后，redis按照正常RDB快照规则生成快照文件，在两次快照期间则通过AOF追加日志到RDB数据后面，因此最终生成的AOF文件是一个以RDB数据开头，多个日志操作结尾的文本文件，新产生的AOF文件会直接覆盖旧的AOF文件。优点：兼具了RDB加载快、数据量小和AOF数据高安全性（日志）的优点，而且也不需要重写过程了。缺点：redis4之前的版本不识别混合AOF文件。

13、Redis的数据类型？

String、hash、list、set、sorted set

14、Spark的RDD、算子分类？

14.1、RDD

RDD是一个弹性分布式数据集，是一个只读的分区记录的集合，只能基于某个数据集或其他RDD上转换而来，因此具有高容错、低开销的特点。

14.2、算子分类

14.2.1、Transformation（转换）算子

map

filter

flatMap

mapPartitions

distinct

sortby

join

14.2.2、Action（行动）算子

reduce

collect

count

first

take

foreach

15、Kafka的分区分配策略？

1）random（随机）：随机分配数据，会导致数据倾斜

2）RoundRobin（轮询）：将Topic中的每个分区以轮询的方式发送给消费者线程进行消费

3）指定分区

16、Kafka丢不丢数据（Ack应答机制）？

我们公司设置ack为-1，并且设置ISR队列中必须至少有两个kafka服务节点（默认ISR队列只有一个Leader，因为这种情况相当于ack等于1），以此来保证可靠性

ack = 0 ，生产者发送完数据就不管了，数据可能会丢

ack = 1，生产者发送数据，Leader也接收到了数据，写入了磁盘，但还未来得及同步数据到其它节点，Leader就宕机了，此时数据会丢

ack = -1，生产者发送数据，Leader接收数据并写入磁盘，也进行了副本同步，若此时Leader宕机，可能发生重复消费，数据可能会重复

17、Kafka的ISR作用（副本同步策略）？

ISR中包括Leader和Follower，若Leader进程挂掉，controller会在ISR队列中选择一个服务作为新的Leader（该队列的作用就是选举老大）

有两个参数决定一台kafka服务是否能加入ISR队列：延迟时间和延迟条数。在0.10版本移除了延迟条数，防止kafka服务频繁的进出ISR队列

18、数仓的数据模型？***

星型模型

数仓（具体说是dwd层）中只有一张包含历史数据且不冗余的事实表和一组附属维度表，每个维度一张。事实表与维度表之间通过外键和主键关联。星型模型的维度表可能存在冗余，因此是反三范式的，这种模型在数据维护上较麻烦，但是性能更高，业界普遍使用星型模型。星型模型的难点在于拉链表的维护，拉链表一般不能有冗余。

雪花模型

针对星型模型的维度表进行扩展的模型，将维度表拆解成维度表+说明表，说明表又可以进一步拆分，最终形成事实表-维度表-说明表的多次连接。雪花模型的表一般遵循三范式，在数据的维护上会很方便，但是多表join影响性能。

星系模型

多个事实表采用星型模型共享维度表，就形成了一个星系。

Data Vault模型

从上述模型中我们不难看出，如果解决了拉链表的维护问题，星型模型的缺陷就已经可以忽略。 DataVault模型由中心表、链接表、附属表、PIT表组成，这里的中心表，事实上就是维度表，链接表就是事实表、附属表是拉链表。 Data Vault模型就是通过将拉链表从维度表中分离出来，来达到方便维护的目的。首先，中心表是一组业务生命周期内绝不会变的维度表，打个比方就是java里头的常量，常量再怎么冗余也不会有影响。链接表则是保存流水类数据的事实表，它通过外键与多张中心表连接，但不会连接附属表。附属表则是从维度表中抽出来的可能会发生变化的字段或表，这一部分就采用拉链表的方式创建，中心表则通过外键关联附属表的主键。从同一维度表拆出来的字段根据变化维度不同可能还要分成多表存储，为了避免时效不一致，所以还会建立一张PIT表，用于维护附属表的变化历史。中心表通过外键与PIT表相连，而PIT表则为每个附属表的主键都准备了一个时间字段保存数据的更新时间。

19、数仓架构有哪些？分别说明 ***

19.1、inmon架构

自上而下的开发模式，从多个数据源出发，根据需求将不同数据源的数据经过ETL过程获取到各个主题需求的数据集成到数仓中，完成了数据治理后再进行统计业务，将统计结果存入数据集市。

19.2、kimball架构

自下而上的开发模式，往往已经存在某个关系明确的业务数据库，架构师需要根据数据库中的数据寻找出有价值的分析指标，然后根据这些指标建立数据集市，再从数据集市出发向下建设需要的数据仓库表。

20、Spark的内存模型有哪些？分别说明

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-fN1gZ6VN-1664237369873)(C:\Users\25851\AppData\Roaming\Typora\typora-user-images\image-20220926105022221.png)]

executor的内存分为4+1块：

Execution:计算用内存，用于执行各种算子时存放临时对象的内存

Storage:缓存用内存，主要存储catch到内存中的数据,广播变量也存在这里

User Memory:用户用内存，存储RDD依赖关系等RDD的信息

Reserved Memory:预留内存，用来存储Spark自己的对象

Offff-heap Memory:堆外内存，开启之后计算和缓存的内存都分别可以存在堆外内存。堆外内存不受spark GC的影响。

Execution和Storage采用联合内存机制，可以互相借用对方的内存区域，但是Execution可以强制征收Storage的内存，反过来不行。

Task共用executor的内存区域，spark准备了一个hashMap用来记录各个task使用的内存，task申请新的内存时，如果剩余内存不够则会阻塞直到有足够的内存为止。每个task至少需要1/2N的内存才能被启动。

21、Kafka数据积压怎么办？

1）提升消费者组中的消费者数以及Topic中的分区数，让二者相等，我们公司设置为3个分区 =3CPU

2）提高消费者拉取数据的能力，比如Flume每次拉取的数据可以由1000条改为3000条、Spark中将限流的参数增大、Flink中保证数据的处理效率等

22、Kafka中的数据是有序的吗？

单分区内有序，分区间无序。如果想让数据有序，可以在生产者生产数据时指定key，让相同的key进入同一个分区中，因为单分区内有序，所以我们在实际开发中用库名+表名作为key值，以此来保证一张表中的数据有序

23、Spark数据倾斜调优方案？

23.1、先用sample(false,0.x)采样key，找出倾斜的key

23.2、把数据集拆成倾斜的部分和不倾斜的部分，不倾斜的部分走正常流程

23.3、倾斜的部分key前面加上一个定长的随机字符串，然后执行重分区

23.4、重分区后进行一个聚合操作，然后去除定长前缀再聚合一次。

23.5、如果是大表join大表，其中一个表有数据倾斜，就需要用膨胀法，将倾斜部分的key加上一个0-n的前缀，一条数据膨胀成n条，然后将另一个表的这部分key也加上相应的前缀，然后单独对这部分数据进行一次双重聚合，与不倾斜的数据进行union操作，完成聚合。

23.6、空值看作是特殊的key，空值多了一样用3的方法去解决。

24、SparkSQL的DF和DS是什么？

DataFrame = DataSet[Row] DataSet是强类型JVM object的集合，DataFrame则是由每行弱类型的JVM object组成，二者都是按列存储，前者是结构化数据集，后者则是半结构化数据集

25、SparkSQL的自定义函数？

spark支持三种自定义函数，UDF、UDAF（用户自定义聚合函数）、UDTF（用户自定义生成函数）

UDAF和UDTF都需要继承对应的自定义函数类，实现相应的抽象方法才可以使用

UDF则可以在spark.udf.register方法中使用函数直接注册使用。

26、SparkShuffle有哪些？

26.1、SortShuffle

mapTask将map（聚合算子）或array（join算子）写入内存
达到阀值发生溢写，溢写前根据key排序，分批写入磁盘，最终将所有临时文件合并成一个最终文件，并建立一份索引记录分区信息。一个mapTask最终形成一个文件。
reduceTask拉取各个task中自己的分区数据去计算。

27、Hive的文件存储格式有那些？区别是什么？

一般选择ORC+bzip/gzip作为数据源的存储，选择则ORC+Snappy作为中间数据的存储分区表单文件不大可以采用gzip压缩，桶表需要用bzip或lzo支持分片的方式压缩

28、Hive的开窗排名函数有哪些？

28.1、开窗函数

sum(col) over() : 分组对col累计求和，over() 中的语法如下 
count(col) over() : 分组对col累计，over() 中的语法如下 
min(col) over() : 分组对col求最小 
max(col) over() : 分组求col的最大值 
avg(col) over() : 分组求col列的平均值first_value(col) over() : 某分区排序后的第一 个col值
last_value(col) over() : 某分区排序后的最后一个col值 
lag(col,n,DEFAULT) : 统计往前n行的col值，n可选，默认为1，DEFAULT当往上第n行为NULL时 候，取默认值，如不指定，则为NULL 
lead(col,n,DEFAULT) : 统计往后n行的col值，n可选，默认为1，DEFAULT当往下第n行为NULL 时候， 取默认值，如不指定，则为NULL 
ntile(n) : 用于将分组数据按照顺序切分成n片，返回当前切片值。注意：n必须为int类型。

28.2、排名函数

row_number() over() : 排名函数，不会重复，适合于生成主键或者不并列排名
rank() over() : 排名函数，有并列名次，名次不连续。如:1,1,3 dense_rank() over() : 排名函数，有并列名次，名次连续。如：1，1，2

row_number 不并列不跳过: 1 2 3 4

dense_rank 并列不跳过: 1 1 2 3

rank 并列跳过: 1 1 3 4

29、Hive的分区和分桶关系？

29.1、分区表

将数据按照分区字段拆分存储的表，在hdfs中以文件夹的形式分别存放不同分区的数据，可以避免全表查询，提高查询效率。

29.2、分桶表

根据分桶字段hash值分组拆分数据的表，在hdfs中表现为将单个的数据文件拆分为多个文件。

29.3、总结

分区字段的每个值都对应一个文件夹和一个分区文件，而分桶字段则是多个值对应一个桶文件。如果同时使用分区和分桶，则会先按照分区划分文件，再对每个文件按照分桶进行拆分。

30、Hive 调优有哪些？***

减少distinct：使用distinct容易造成数据倾斜问题，使用group by的子查询代替它。

map任务数量优化：

实际业务中往往存在大量的分区表，每个分区表都实际存储一定量的文件，其中必然有些分区的数据量很少。正常读取时往往有多少个文件就创建多少个map，此时可以通过设置一些参数，让sql语句在执行前先合并表文件。

-参数：

mapred.min.split.size.per.node = {设置一个节点中分片至少的大小}byte

mapred.min.split.size.per.rack= {设置一个交换机中分片至少的大小}byte mapred.max.split.size ={设置分片的最大大小}byte

hive.input.format=org.apache.hadoop.hive.ql.io.CombineHiveInputFormat

设置hive先合并文件再执行

-效果：假设我全部设置为100000000（相当于100M），经过配置后，hive会首先合并文件，切分成各种100M，最后再把剩下来的各个节点上的散碎数据合并到一起再生成几个分片。还有一种情况，当一个map任务中处理数据量很大时（大小很小，但是条数很多），可以采用分桶法，先用一个查询语句把该表数据查出来分桶写入，再使用这个分桶表。相当于增加map任务数量，增加并行度。

并行度优化：

1.手动设置reduce数量 mapred.reduce.tasks

2.避免全局的聚合函数，使用聚合函数尽量要分组

3.避免全局的order by，有时候全局排序很难避免，但可以根据topN需求，再各个分区中只留下N个值，再进行全局排序。

4.避免笛卡尔积 5.设置mapTask分片大小

小文件问题：

1.避免产生小文件：少用动态分区、根据需求使用reduce

2.使用Sequencefifile作为表存储格式，不要用textfifile，在一定程度上可以减少小文件

3.使用hadoop archive命令把小文件进行归档

4.重建表，建表时减少reduce数量 5.参数设置： hive.merge.mapfifiles = true 设置map端输出合并

hive.merge.mapredfifiles = true

设置reduce输出合并

hive.merge.size.per.task = 25610001000

设置合并文件的大小

hive.merge.smallfifiles.avgsize=16000000

设置当平均大小小于该值时合并

存储格式

1.使用ORCfifile存储，可以显著提高join操作的查询速度 2.使用压缩格式存储，可以显著降低网络IO和存储大小

使用map端join

使用tez作为默认引擎

使用向量化查询：

一次执行1024行数据的操作 hive.vectorized.execution.enabled = true;

hive.vectorized.execution.reduce.enabled = true;

设置本地模式、并行模式（自动并行非依赖阶段）、严格模式

开启JVM重用

总结

减少distinct 
设置读取时合并小文件和合理拆分大文件 
优化并行度 
设置存储格式和压缩格式 
设置输出时合并小文件 
设置map端JOIN 
更换引擎 
设置本地模式、并行模式、严格模式 
开启JVM重用 开启推测执行

31、Kafka的Segment是什么？里面包含什么？

32、Kafka架构？

生产者、Broker主机节点、消费者、Zookeeper保存Broker id和消费者offffsets信息

33、Spark的分区分类，如何使用自定义分区？

hive通过设置hive.exec.dynamici.partition=true开启动态分区。

可以在插入数据时根据表中某字段值决定分区，当分区字段完全由变量决定时称为动态分区，若有常量限制则称为混合分区，若完全由常量决定分区时称为静态分区。

自定义分区器

通过实现get分区总数方法和get分区数方法，指定自定义规则的key分区方式; 使用自定义分区器创建的RDD进行复杂的聚合或join操作效率更高。

34、Spark的DAG是什么，说明其功能？

\1. sparkContext创建DAGScheduler->创建EventProcessLoop->调用eventLoop.start()方法开启事件监听

\2. action调用sparkContext.runJob->eventLoop监听到事件，调用handleJobSubmitted开始划分stage

\3. 首先对触发job的fifinalRDD调用createResultStage方法，通过getOrCreateParentStages获取所有父stage列表，然后创建自己。

如：父(stage1,stage2)，再创建自己stage3

\4. getOrCreateParentStages内部会调用getShufflfflffleDependencies获取所有直接宽依赖（从后往前推，窄依赖直接跳过）在这个图中G的直接宽依赖是A和F，B因为是窄依赖所以跳过，所以最后B和G属于同一个stage

\5. 接下来会循环宽依赖列表，分别调用getOrCreateShufflfflffleMapStage：

– 如果某个RDD已经被划分过会直接返回stageID；否则就执行

getMissingAncestorShufflfflffleDependencies方法，继续寻找该RDD的父宽依赖，窄依赖老规矩直接加入:

– 如果返回的宽依赖列表不为空，则继续执行4,5的流程直到为空为止;

– 如果返回的宽依赖列表为空,则说明它没有父RDD或者没有宽依赖，此时可以直接调用createShufflfflffleMapStage将该stage创建出来

\6. 因此最终的划分结果是stage3(B,G)、stage2(C,D,E,F)、stage1(A)

\7. 创建ResultStage，调用submitStage提交这个stage

\8. submitStage会首先检查这个stage的父stage是否已经提交，如果没提交就开始递归调用submitStage提交父stage，最后再提交自己。

\9. 每一个stage都是一个taskSet，每次提交都会提交一个taskSet给TaskScheduler

35、ClickHouse表引擎有哪些，详细介绍一下MergeTree引擎

36、ClickHouse位图函数和行列转换函数分别是什么？

37、ClickHouse的数据类型有哪些？

38、Zookeeper选举机制？

38.1 容错机制（为什么单数节点）

剩下的节点数必须大于宕掉的个数，也就是存货节点需要大于总数一半，zookeeper才可以继续使用。

假设总结点为5，则需要3台存活，即使增加为6也是3台，也就是说只有节点数为增加到奇数时最小存活节点才会增加，所以设置节点为奇数节约资源。

38.2 选举机制（重点）

暂停对外服务，

\1. 各节点会先选自己作为leader，然后将选票携带事务id:zxid发送出去

\2. 各节点拿到选票后，先排除非本轮的票，然后比对自己的选票跟各个节点发来的选票，先比较zxid（越大说明数据越新），相同时比较myid，大的一方获胜，将票投给获胜方，然后各自发回节点。（假如自己是1节点，3节点发来的票，比对后要么返回1要么返回3）

\3. 投票后，各个节点会统计投票信息，判断如果有过半选票则认为选出了leader，更新自身状态为follow或leader，如果没有则一直重复2直到满足条件为止。

\4. 选举出leader后，新节点或原leader节点宕机恢复后，会直接变为follow状态，不再进行选举。

39、SparkOnHive与HiveOnSpark如何实现，区别是什么？

40、Spark On Yarn工作原理？

41、Spark任务提交工作原理？

41.1、 standalone

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-wgJtCdZX-1664237369874)(C:\Users\25851\AppData\Roaming\Typora\typora-user-images\image-20220926140316596.png)]

\1. driver端:通过反射获取主类执行main方法 -> 创建sparkconf和sparkContext，创建通信环境、后端调度器（负责向master发送注册信息、向excutor发送task的调度器）、task调度器、DAG（根据宽窄依赖划分stage）调度器 ->封装任务信息提交给Master

\2. Master端:缓存任务信息并将其放入任务队列 -> 轮到该任务时，调用调度方法进行资源调度 ->发送调度信息给对应的worker

\3. Worker端:worker将调度信息封装成对象 -> 调用对象的start方法，启动excutor进程

\4. Excutor进程:启动后向driver端反向注册(driver端拿到信息后注册excutor，向其发送任务) -> 创建线程池，封装任务对象 -> 获取池中线程执行任务 -> 反序列化TastSet，执行给定的各种算子步骤

41.2、yarn-client

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g5dAlAQT-1664237369874)(C:\Users\25851\AppData\Roaming\Typora\typora-user-images\image-20220926140630945.png)]

\1. 客户端向yarn的RM申请启动AM，同时在自身的sparkContext中创建DAGScheduler和 TASKScheduler（创建driver）

\2. 按照正常Yarn流程，一个NM领取到AM任务作为AM与客户端的driver产生连接（在yarn-cluster中该AM直接作为driver而不是连接driver）

\3. driver根据任务信息通过AM向RM申请资源（计算容器）

\4. AM通知领取到任务的NM向driver的sparkContext反注册并申请Task

\5. driver的sparkContext分配Task给各个计算节点，并随时掌握各个任务运行状态

\6. 应用程序运行完成后，sparkContext向RM申请注销并关闭自己。

总结：与standalone区别是，AM只作为中间联系，实际作为AM的是driver的sparkContext

41.3、yarn-cluster

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-j3vJMoyo-1664237369874)(C:\Users\25851\AppData\Roaming\Typora\typora-user-images\image-20220926140832761.png)]

\1. 先将driver作为一个AM在一个NM中启动

\2. 由AM创建应用程序，走正常的yarn流程启动Executor运行Task，直到运行完成

总结：与yarn client相比只是把driver端由客户端变成了集群中的某个NodeManager节点。

42、JVM的GC是什么？

JVM内存按回收机制可分为年轻代和老年代，年轻代分为eden区和多个幸存区，老年代则不分区。

无论是YGC还是Full GC，都会使java线程暂停，但是YGC暂停的事件极短，因此基本是针对减少Full GC的方向优化。可以通过参数指定分配的最大堆大小、初始堆大小、年轻代大小、比值、指定并发收集器、并行收集器等等。

42.1 YGC

新的对象会存入eden区，当eden区满了放不下的时候，会对年轻代的内存进行垃圾回收，eden中有用的对象移到幸存区，清空eden区，称为YGC。

42.2 Full GC

某个对象经过多次YGC后依然存活，会移植到老年代。当老年代满了放不下的时候，就会触发FullGC，对整个内存进行一次垃圾回收。无论是YGC还是Full GC，都会使java线程暂停，但是YGC暂停的事件极短，因此基本是针对减少Full GC的方向优化。可以通过参数指定分配的最大堆大小、初始堆大小、年轻代大小、比值、指定并发收集器、并行收集器等等。

42.3 垃圾回收机制（算法原理）

引用计数法：对象每被引用一次就+1，为0时回收，速度很快但是无法识别循环引用

标记清除法：遍历所有对象，标记没被引用的，然后统一清除。缺点是效率低、清理后内存不连续。

复制清除法：将内存分为两块，其中一块写满后，遍历对象标记有用的对象复制到另一块，然后把这一块清理，这样复制的内容很少而且内存始终连续，缺点是始终需要有一块内存空出来用于复制。

标记整理法：遍历出有用的对象，将这些对象全都向一端移动，然后清理其它空间，一样能腾出连续的内存，但是移动对象的成本比复制大得多。

GC采用分代收集法：年轻代采用复制清除法，每当eden满时，就遍历出eden和幸存者1区的有用对象复制到幸存者2区，然后清空重新写起。因此无论何时一定有一个幸存者区是空的。老年代由于有用的对象很多所以复制成本高，采用标记整理法减少复制。

43、数据仓库的数据发散如何产生的？如何解决？

44、Flink 状态存储？

Flink在做计算的过程中经常需要存储中间状态，来避免数据丢失和状态恢复。选择的状态存储策略不同，会影响状态持久化如何和 checkpoint 交互。Flink提供了三种状态存储方式：MemoryStateBackend、FsStateBackend、RocksDBStateBackend。

45、Flink 窗口？

Flink 支持两种划分窗口的方式，按照time和count。如果根据时间划分窗口，那么它就是一个time-window 如果根据数据划分窗口，那么它就是一个count-window。
flink支持窗口的两个重要属性（size和interval）如果size=interval,那么就会形成tumbling-window(无重叠数据)如果size>interval,那么就会形成sliding-window(有重叠数据)如果size< interval, 那么这种窗口将会丢失数据。比如每5秒钟，统计过去3秒的通过路口汽车的数据，将会漏掉2秒钟的数据。

通过组合可以得出四种基本窗口：

time-tumbling-window 无重叠数据的时间窗口，设置方式举例：timeWindow(Time.seconds(5))
time-sliding-window 有重叠数据的时间窗口，设置方式举例：timeWindow(Time.seconds(5),Time.seconds(3))
count-tumbling-window无重叠数据的数量窗口，设置方式举例：countWindow(5)
count-sliding-window 有重叠数据的数量窗口，设置方式举例：countWindow(5,3)

46、Flink State分类？

State：指一个具体的Task/Operator的状态。State可以被记录，在失败的情况下数据还可以恢复，Flink中有两种基本类型的State： Keyed State ， Operator State 。Sate在数据在容错恢复起了非常关键的作用

Operator State

Task里面没有Shufflfflffle操作的State，换句话说，就是没有keyBy操作

\1. Operator State是Task级别的state，就是每个Task对应一个State

\2. Kafka Connector Source中的每个分区（Task）都需要记录消费的Topic的Partition和Offffset等信息，这些信息就是state。

Keyed State

表示和Key相关的一种State，基于KeyedStream上的状态。Keyed State事先按照Key对数据集进行了分区，每个Key State仅对应一个Operator和Key的组合。Keyed State可以通过Key Groups进行管理，主要用于当算子并行度发生变化时，自动重新分布Keyed State数据。在系统运行过程中，一个Keyed算子实例可能运行一个或者多个Key Groups 的 Keys

47、Flink checkpoint 和savepoint 区别？

Checkpoint 是为 runtime 准备的，Savepoint 是为用户准备的。Checkpoint 机制的目标在于保证Flink 作业意外崩溃重启不影响 exactly once 准确性，通常用于系统容错。而 Savepoint 的目的在于在Flink 作业维护（比如更新作业代码）时将作业状态写到外部系统，以便维护结束后重新提交作业可以到恢复原本的状态。

Checkpoint 异常恢复，保证可用性，在任务发生故障时，为任务提供给自动恢复机制；Savepoint 需手动备份、恢复暂停作业的方法。
Checkpoint 被设计成轻量和快速恢复数据的机制，Savepoint 更多地关注数据的可移植性，并支持对作业做任何更改而状态能保持兼容
Checkpoint 是自动和定期的，它们由 Flink 自动地周期性地创建和删除，无需用户的交互。相反，Savepoint 是由用户手动地管理（调度、创建、删除）的。

48、Flink的延迟数据如何处理？有哪些方案？

49、Flink 如何处理乱序事件？

Flink 使用Watermark 机制来处理乱序事件。我们知道，流处理从事件产生，到流经source，再到operator，中间是有一个过程和时间的。虽然大部分情况下，流到operator的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、背压等原因，导致乱序的产生(out-of-order或者说late element),但是对于late element，我们又不能无限期的等下去，必须要有个机制来保证一个特定的时间后，必须触发window去进行计，这个机制，就是watermark，它本质上就是一个时间戳。

# Watermark 如何工作的 
1. 可以把Watermarks理解为一个水位线，这个Watermarks在不断的变化。Watermark实际上作为数 据流的一部分随数据流流动。 
2. Watermark是一种告诉Flink一个消息延迟多少的方式。它定义了什么时候不再等待更早的数据 
3. 当Flink中的运算符接收到Watermarks时，它明白早于该时间的消息已经完全抵达计算引擎，即假 设不会再有时间小于水位线的事件到达 
4. 这个假设是触发窗口计算的基础，只有水位线越过窗口对应的结束时间，窗口才会关闭和进行计算。

Watermark 设定方式

\1. Periodic Watermark

周期性的（允许一定时间间隔或者达到一定的记录条数）产生一个Watermark。水位线提升的时

间间隔是由用户设置的，在两次水位线提升时隔内会有一部分消息流入，用户可以根据这部分数据

来计算出新的水位线

\2. Punctuated Watermark

标点水位线（Punctuated Watermark）通过数据流中某些特殊标记事件来触发新水位线的生成。

这种方式下窗口的触发与时间无关，而是决定于何时收到标记事件

50、Flink的Slot和Parallelism是什么？有什么关系？

Slot 是指 TaskManager 的并发执行能力，如果代码运行前我们将 slot 的个数配置为3（taskmanager.numberOfTaskSlots），那么每个 TaskManager 会分配 3 个 Slot 来执行 task，如果配置了 3 个 taskmanager 那么就如图一共有 9 个 Slot。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DK4103kD-1664237369875)(C:\Users\25851\AppData\Roaming\Typora\typora-user-images\image-20220926142515041.png)]

Parallelism 是指 TaskManager 在实际运行过程中的并发。默认并行度的配置为1（parallelism.default），那么如图 9 个 Slot 只有一个是在工作的，其他 8 个都空闲。在用户开发的过程中可以通过 setParallelism 方法给每个 Operators 算子配置并行度。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-B35iHwNG-1664237369875)(C:\Users\25851\AppData\Roaming\Typora\typora-user-images\image-20220926142610779.png)]

51、Flink的分布式缓存是什么？

Flink 提供了一个分布式缓存，类似于 hadoop，可以使用户在并行函数中很方便的读取本地文件，并把它放在 taskmanager 节点中，防止 task 重复拉取。

此缓存的工作机制如下：程序注册一个文件或者目录(本地或者远程文件系统，例如 hdfs 或者 s3)，通过 ExecutionEnvironment 注册缓存文件并为它起一个名称。

当程序执行，Flink 自动将文件或者目录复制到所有 taskmanager 节点的本地文件系统，仅会执行一次。用户可以通过这个指定的名称查找文件或者目录，然后从 taskmanager 节点的本地文件系统访问它。

52、Flink On Yarn分类，分别说明区别？

Flink的yarn部署模式分为三种方式:一种是Application Mode，一种是Per-Job Mode，一种是Session Mode(注：flink1.9及以前只有后两种)。

Flink 应用程序的作业可以被提交到长期运行的 Flink Session 集群、专用的 Flink Job 集群或 Flink Application 集群

**Application Mode：**针对提交的Job启动一个专用的JobManager，这个JM将仅仅执行这个作业，完成后退出；Flink Application是运行在JobManager中(你也可以认为client程序都在JM中运行)。

**Per-Job Mode：**针对提交的Job启动一个专用的JobManager，这个JM将仅仅执行这个作业，完成后退出；Flink Application是运行在提交该应用的client所在服务器上。

**Session Mode：**多个job共用一个JobManager，JM一直不退出；Flink Application是运行在提交该应用的client所在服务器上。

ession 集群](https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/concepts/glossary.html#flink-session-cluster)、专用的 Flink Job 集群或 Flink Application 集群

**Session Mode：**多个job共用一个JobManager，JM一直不退出；Flink Application是运行在提交该应用的client所在服务器上。