大数据之Spark:Spark面试（中级）

Posted 2022-06-04 浊酒南街

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了大数据之Spark:Spark面试（中级）相关的知识，希望对你有一定的参考价值。

11：如何理解DAGScheduler的Stage划分算法
12：如何理解TaskScheduler的Task分配算法
13：Spark的本地化级别有哪几种？怎么调优
14：说说Spark和Mapreduce中Shuffle的区别
15：Spark的内存是怎么管理的
16：Spark的广播变量和累加器的作用是什么
17：Spark SQL和Hive SQL的区别
18：说下Spark SQL的执行流程

11：如何理解DAGScheduler的Stage划分算法

首先放上官网的RDD执行流程图:

针对一段应用代码(如上)，Driver会以Action算子为边界生成DAG调度图。DAGScheduler从DAG末端开始遍历划分Stage，封装成一系列的tasksets移交TaskScheduler，后者根据调度算法, 将taskset分发到相应worker上的Executor中执行。

DAGSchduler的工作原理

DAGScheduler是一个面向stage调度机制的高级调度器，为每个job计算stage的DAG(有向无环图)，划分stage并提交taskset给TaskScheduler

追踪每个RDD和stage的物化情况，处理因shuffle过程丢失的RDD，重新计算和提交。

查找rdd partition 是否cache/checkpoint。提供优先位置给TaskScheduler，等待后续TaskScheduler的最佳位置划分

Stage划分算法

从触发action操作的算子开始，从后往前遍历DAG

为最后一个rdd创建finalStage

遍历过程中如果发现该rdd是宽依赖，则为其生成一个新的stage，与旧stage分隔而开，此时该rdd是新stage的最后一个rdd

如果该rdd是窄依赖，将该rdd划分为旧stage内，继续遍历

以此类推，继续遍历直至DAG完成

12：如何理解TaskScheduler的Task分配算法

TaskScheduler负责Spark中的task任务调度工作。TaskScheduler内部使用TasksetPool调度池机制存放task任务。TasksetPool分为FIFO(先进先出调度)和FAIR(公平调度)。

FIFO调度: 基于队列思想，使用先进先出原则顺序调度taskset

FAIR调度: 根据权重值调度，一般选取资源占用率作为标准，可人为设定

TaskScheduler的工作原理

负责Application在Cluster Manager上的注册

根据不同策略创建TasksetPool资源调度池，初始化pool大小

根据task分配算法发送Task到Executor上执行

Task分配算法

首先获取所有的executors，包含executors的ip和port等信息

将所有的executors根据shuffle算法进行打散

遍历executors。在程序中依次尝试本地化级别，最终选择每个task的最优位置(结合DAGScheduler优化位置策略)

序列化task分配结果，并发送RPC消息等待Executor响应

13：Spark的本地化级别有哪几种？怎么调优

移动计算 or 移动数据？这是一个问题。在分布式计算的核心思想中，移动计算永远比移动数据要合算得多，如何合理利用本地化数据计算是值得思考的一个问题。

TaskScheduler在进行task任务分配时，需要根据本地化级别计算最优位置，一般是遵循就近原则，选择最近位置和缓存。Spark中的本地化级别在TaskManager中定义，分为五个级别。

Spark本地化级别

PROCESS_LOCAL(进程本地化)

partition和task在同一个executor中，task分配到本地Executor进程。

NODE_LOCAL(节点本地化)

partition和task在同一个节点的不同Executor进程中，可能发生跨进程数据传输

NO_PREF(无位置)

没有最佳位置的要求，比如Spark读取JDBC的数据

RACK_LOCAL(机架本地化)

partition和task在同一个机架的不同worker节点上，可能需要跨机器数据传输

ANY(跨机架): 数据在不同机架上，速度最慢

Spark本地化调优

在task最佳位置的选择上，DAGScheduler先判断RDD是否有cache/checkpoint，即缓存优先；否则TaskScheduler进行本地级别选择等待发送task。

TaskScheduler首先会根据最高本地化级别发送task，如果在尝试5次并等待3s内还是无法执行，则认为当前资源不足，即降低本地化级别，按照PROCESS->NODE->RACK等顺序。

调优1：加大spark.locality.wait 全局等待时长

调优2：加大spark.locality.wait.xx等待时长(进程、节点、机架)

调优3：加大重试次数(根据实际情况微调)

14：说说Spark和Mapreduce中Shuffle的区别

Spark中的shuffle很多过程与MapReduce的shuffle类似，都有Map输出端、Reduce端，shuffle过程通过将Map端计算结果分区、排序并发送到Reducer端。

下面将对Spark和Mapreduce中shuffle过程分开叙述，Mapreduce的shuffle大家都不陌生了，主要重点突出Spark的Shuffle机制做了哪些优化工作。

Hadoop Mapreduce Shuffle

MapReduce的shuffle需要依赖大量磁盘操作，数据会频繁落盘产生大量IO，同时产生大量小文件冗余。虽然缓存buffer区中启用了缓存机制，但是阈值较低且内存空间小。

读取输入数据，并根据split大小切分为map任务

map任务在分布式节点中执行map()计算

每个map task维护一个环形的buffer缓存区，存储map输出结果，分区且排序

当buffer区域达到阈值时，开始溢写到临时文件中。map task任务结束时进行临时文件合并。此时，整合shuffle map端执行完成

mapreduce根据partition数启动reduce任务，copy拉取数据

merge合并拉取的文件

reduce()函数聚合计算，整个过程完成

2. Spark的Shuffle机制

Spark1.2以前，默认的shuffle计算引擎是HashShuffleManager，此种Shuffle产生大量的中间磁盘文件，消耗磁盘IO性能。在Spark1.2后续版本中，默认的ShuffleManager改成了SortShuffleManager，通过索引机制和合并临时文件的优化操作，大幅提高shuffle性能。

HashShuffleManager

HashShuffleManager的运行机制主要分成两种，一种是普通运行机制，另一种是合并的运行机制。合并机制主要是通过复用buffer来优化Shuffle过程中产生的小文件的数量，Hash shuffle本身不排序。开启合并机制后，同一个Executor共用一组core，文件个数为cores * reduces。

SortShuffleManager

SortShuffleManager的运行机制分成两种，普通运行机制和bypass运行机制。当shuffletask的数量小于等于spark.shuffle.sort.bypassMergeThreshold参数的值时(默认200)，会启用bypass机制。

SortShuffleManager机制采用了一个特殊的内存数据结构(Map)，数据优先写入此结构中，当达到阈值时溢写到磁盘中并清空内存数据结构。在过程中对数据进行排序并合并，减少最终的临时文件数量。ByPass机制下在其基础上加了一个索引机制，将数据存放位置记录hash索引值，相同hash的数据合并到同一个文件中。

15：Spark的内存是怎么管理的

Spark内存分为堆内内存和堆外内存，其中堆内内存基于JVM实现，堆外内存则是通过调用JDK Unsafe API管理。在Spark1.6版本前后内存管理模式分为: 静态管理(Static Memory)和统一管理(Unified Memory)。

16：Spark的广播变量和累加器的作用是什么

Executor接收到TaskScheduler的taskset分发命令，根据rdd分区数在ThreadPool中创建对应的Task线程，每个Task线程拉取并序列化代码，启动分布式计算。

Spark在计算过程中的算子函数、变量都会由Driver分发到每台机器中，每个Task持有该变量的一个副本拷贝。可是这样会存在两个问题: