Spark Core基础面试题总结（下）

大数据真有意思 2020-10-19

140

点击关注上方“知了小巷”，

设为“置顶或星标”，第一时间送达干货。

【Spark Core基础面试题总结（上）】

16. Spark Lineage

Spark血统是Spark在处理分布式计算环境下的数据容错性问题时采用方案的一部分。数据容错：一般包括节点失效、数据丢失。
Lineage：用来描述不同RDD之间的依赖关系。

为了保证RDD中数据的鲁棒性【抵御或克服不利条件或出错的能力】，RDD数据集通过血缘关系Lineage记住了它是如何从其它RDD中演变过来的。相比其它系统的细颗粒度的内存数据更新级别的备份或者Log机制。RDD的Lineage记录的是粗粒度的特定数据转换（Transformation）操作（包括filter map join等等）。当这个RDD的部分分区数据丢失时，它可以通过Lineage获取足够的信息来重新计算和恢复丢失的数据分区。这种粗颗粒的数据模型，限制了Spark的应用场景，但同时也带来了相比粗粒度的数据模型更好的性能提升。

RDD在Lineage依赖方面包括两种：窄依赖和宽依赖，用来解决数据容错时的高效性。

/**
 * :: DeveloperApi ::
 * Base class for dependencies.
 */
@DeveloperApi
abstract class Dependency[T] extends Serializable {
  def rdd: RDD[T]
}

// ShuffleDependency 宽依赖
@DeveloperApi
class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
    @transient private val _rdd: RDD[_ <: Product2[K, V]],
    val partitioner: Partitioner,
    val serializer: Serializer = SparkEnv.get.serializer,
    val keyOrdering: Option[Ordering[K]] = None,
    val aggregator: Option[Aggregator[K, V, C]] = None,
    val mapSideCombine: Boolean = false,
    val shuffleWriterProcessor: ShuffleWriteProcessor = new ShuffleWriteProcessor)
  extends Dependency[Product2[K, V]] { //...}

// NarrowDependency 窄依赖 
// Allow for pipelined execution.
@DeveloperApi
abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
  /**
   * Get the parent partitions for a child partition.
   * @param partitionId a partition of the child RDD
   * @return the partitions of the parent RDD that the child partition depends upon
   */
  def getParents(partitionId: Int): Seq[Int]

  override def rdd: RDD[T] = _rdd
}  

// 窄依赖 OneToOneDependency
@DeveloperApi
class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency[T](rdd) {
  override def getParents(partitionId: Int): List[Int] = List(partitionId)
}

// 窄依赖 RangeDependency
@DeveloperApi
class RangeDependency[T](rdd: RDD[T], inStart: Int, outStart: Int, length: Int)
  extends NarrowDependency[T](rdd) {

  override def getParents(partitionId: Int): List[Int] = {
    if (partitionId >= outStart && partitionId < outStart + length) {
      List(partitionId - outStart + inStart)
    } else {
      Nil
    }
  }
}

// 窄依赖 分支裁剪 PruneDependency
private[spark] class PruneDependency[T](rdd: RDD[T], partitionFilterFunc: Int => Boolean)
  extends NarrowDependency[T](rdd) {

  @transient
  val partitions: Array[Partition] = rdd.partitions
    .filter(s => partitionFilterFunc(s.index)).zipWithIndex
    .map { case(split, idx) => new PartitionPruningRDDPartition(idx, split) : Partition }

  override def getParents(partitionId: Int): List[Int] = {
    List(partitions(partitionId).asInstanceOf[PartitionPruningRDDPartition].parentSplit.index)
  }
}

窄依赖：是指父RDD的每一个分区最多被一个子RDD的分区所使用，也就是父RDD不会被分叉，表现为一个父RDD的分区对应于一个子RDD的分区或多个父RDD的分区对应于一个子RDD的分区，也就是一个父RDD的一个分区不可能对应一个子RDD的多个分区。

宽依赖：是指子RDD的分区依赖于父RDD的多个分区或所有分区，也就是说存在一个父RDD的一个分区对应一个子RDD的多个分区，也就是做了group by分叉或分发。

对于宽依赖，这种计算的输入和输出发生在不同的节点上，Lineage方法对于输入节点保持完好，而输出节点宕机时，通过重新计算，这种情况下，这种方法容错是有效的，否则无效，因为没有办法再重试了，需要向上寻找和追溯到祖先RDD看是否可以重试。窄依赖对于数据的重算开销要远小于宽依赖的数据重算开销

在RDD计算，通过Checkpoint进行容错，做Checkpoint有两种方式：一个是Checkpoint data，一个是logging the updates；就是直接记录数据本身和记录更新操作。用户可以控制采用哪种方式来实现容错，默认是logging the updates方式，通过记录跟踪所有生成RDD的转换（Transformation）也就是记录每个RDD的Lineage来重新计算生成丢失的分区数据。

17. Spark RDD的持久化机制

/**
 * :: DeveloperApi ::
 * Flags for controlling the storage of an RDD. 
 * Each StorageLevel records whether to use memory, or ExternalBlockStore, 
 * whether to drop the RDD to disk if it falls out of memory or ExternalBlockStore, 
 * whether to keep the data in memory in a serialized format, and whether to replicate the RDD partitions on multiple nodes.
 */
@DeveloperApi
class StorageLevel private(
    private var _useDisk: Boolean,
    private var _useMemory: Boolean,
    private var _useOffHeap: Boolean,
    private var _deserialized: Boolean,
    private var _replication: Int = 1)
  extends Externalizable {
  // ...
  def useDisk: Boolean = _useDisk
  def useMemory: Boolean = _useMemory
  def useOffHeap: Boolean = _useOffHeap
  def deserialized: Boolean = _deserialized
  def replication: Int = _replication
  // ... 
}  

// 持久化组合
object StorageLevel {
  val NONE = new StorageLevel(false, false, false, false)
  val DISK_ONLY = new StorageLevel(true, false, false, false)
  val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
  val MEMORY_ONLY = new StorageLevel(false, true, false, true)
  val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
  val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
  val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
  val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
  val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
  val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
  val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
  val OFF_HEAP = new StorageLevel(true, true, true, false, 1)
  // ...
}

cache() 和 persist()
没有参数、默认情况下：
persist: MEMORY_ONLY
cache: MEMORY_ONLY

// 调用persist(StorageLevel.MEMORY_ONLY)
def persist(): this.type = persist(StorageLevel.MEMORY_ONLY)

// 调用persist()
def cache(): this.type = persist()

// 自定义StorageLevel
def persist(newLevel: StorageLevel): this.type = {
  if (isLocallyCheckpointed) {
    // This means the user previously called localCheckpoint(), which should have already
    // marked this RDD for persisting. Here we should override the old storage level with
    // one that is explicitly requested by the user (after adapting it to use disk).
    persist(LocalRDDCheckpointData.transformStorageLevel(newLevel), allowOverride = true)
  } else {
    persist(newLevel, allowOverride = false)
  }
}

当对RDD执行持久化操作时，每个节点都会将自己操作的RDD的partition持久化到内存中，并且在之后对该RDD的反复使用中，直接使用内存缓存的Partition，这样的话，针对一个RDD的反复执行多个操作的场景，就只要对RDD计算一次即可，后面直接使用该RDD，而不需要每次都重新计算RDD。

巧妙地使用RDD持久化，在有些场景下可以将Spark应用的性能提升10倍。对于迭代计算和快速交互式应用来说，RDD持久化，是非常有用的。

要持久化一个RDD，只要调用cache或者persist(StorageLevel)方法即可。在该RDD第一次被计算出来时，就会直接缓存在每个节点中。而且Spark的持久化机制还是自动容错的，如果持久化的RDD的任何Partition丢失了，那么Spark会自动通过它的源RDD，使用Transformation操作重新计算该Partition。

cache和persist的区别如上述源码片段，简化和可变参数的区别。如果要从内存或磁盘中移除RDD缓存，可以使用unpersist(true)。

// Mark the RDD as non-persistent, and remove all blocks for it from memory and disk.
def unpersist(blocking: Boolean = false): this.type = {
  logInfo(s"Removing RDD $id from persistence list")
  sc.unpersistRDD(id, blocking)
  storageLevel = StorageLevel.NONE
  this
}

checkpoint
checkpoint & localCheckpoint

/**
 * Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint
 * directory set with `SparkContext#setCheckpointDir` and all references to its parent
 * RDDs will be removed. This function must be called before any job has been
 * executed on this RDD. It is strongly recommended that this RDD is persisted in
 * memory, otherwise saving it on a file will require recomputation.
 */
def checkpoint(): Unit = RDDCheckpointData.synchronized {
  // NOTE: we use a global lock here due to complexities downstream with ensuring
  // children RDD partitions point to the correct parent partitions. In the future
  // we should revisit this consideration.
  if (context.checkpointDir.isEmpty) {
    throw new SparkException("Checkpoint directory has not been set in the SparkContext")
  } else if (checkpointData.isEmpty) {
    checkpointData = Some(new ReliableRDDCheckpointData(this))
  }
}

// Mark this RDD for local checkpointing using Spark's existing caching layer.
// Local checkpointing sacrifices fault-tolerance for performance. 
def localCheckpoint(): this.type = RDDCheckpointData.synchronized {
  if (conf.get(DYN_ALLOCATION_ENABLED) &&
      conf.contains(DYN_ALLOCATION_CACHED_EXECUTOR_IDLE_TIMEOUT)) {
    logWarning("Local checkpointing is NOT safe to use with dynamic allocation, " +
      "which removes executors along with their cached blocks. If you must use both " +
      "features, you are advised to set `spark.dynamicAllocation.cachedExecutorIdleTimeout` " +
      "to a high value. E.g. If you plan to use the RDD for 1 hour, set the timeout to " +
      "at least 1 hour.")
  }
  // ...
}

使用场景：
当业务场景非常复杂的时候，RDD的Lineage依赖会非常的长，一旦Lineage中靠后的RDD分区数据丢失的时候，Spark会根据Lineage依赖重新计算丢失的RDD数据，这样会造成计算的时间过长，Spark提供了一个叫checkpoint的算子来解决这样的业务场景。

用法：
为当前RDD设置检查点。该函数将会创建一个二进制的文件，并存储到checkpoint目录中，该目录是用SparkContext#setCheckpointDir设置的。在checkpoint的过程中，该RDD的所有依赖于父RDD中的信息将全部被移除。对RDD进行checkpoint操作并不会马上被执行，必须执行Action操作才能触发。

checkpoint的优点：

持久化在HDFS上，HDFS默认的3副本备份使得持久化备份数据更加安全。
切断RDD的依赖关系：当业务场景复杂的时候，RDD的依赖关系非常的长的时候，当比较靠后的RDD数据丢失的时候，会经历较长的重新计算的过程，采用checkpoint会转为依赖CheckpointRDD，可以避免长的Lineage重新计算。
建议checkpoint之前进行cache操作，这样会直接将内存中的结果进行checkpoint，不用重新启动job重新计算。

checkpoint的原理：

当finalRDD执行Action算子计算Job任务的时候，Spark会从finalRDD从后往前回溯查看哪些RDD使用了checkpoint算子。
将使用了checkpoint的算子标记起来。
Spark会自动的启动一个Job来重新计算标记了的RDD，并将计算的结果存入HDFS，然后切断RDD的依赖关系。

18. Spark提交任务的整个流程

ApplicationMaster
在YARN中，每个Application实例都有一个特定于应用客户端的ApplicationMaster进程，也是运行在Container之上。它负责和ResourceManager打交道，并请求资源。获取资源之后告诉NodeManager为其启动container（Executor）。

yarn-cluster和yarn-client模式的区别
yarn-cluster和yarn-client模式的区别其实就是ApplicationMaster(AM)进程的区别，yarn-cluster模式下，driver运行在AM中（一个线程），它负责向YARN申请资源，并监督作业的运行状况。
当用户提交了作业之后，就可以关掉Client，作业会继续在YARN上运行，显然yarn-cluster模式不适合运行交互类型的作业。

而yarn-client模式下，ApplicationMaster仅仅向YARN请求executor，client会和请求的Container通信来调度他们工作，也就是说Client不能离开。

yarn-cluster
与standalone模式不同，yarn-cluster是基于yarn集群进行调度管理的，yarn集群上有ResourceManager（RM）和NodeManager（NM）两个角色。
作业提交流程

由client向RM提交请求，并上传jar到HDFS上。
这期间包括四个步骤：

a). 连接到RM
b). 从RM ASM（Applications Manager）中获得metric、queue和resource等信息。
c). 上传 app jar and spark-assembly jar
d). 设置运行环境和container上下文（launch-container.sh等脚本)

ASM 向 Scheduler 申请空闲 container
Scheduler 向 ASM 返回空闲 container 信息(NM 等)
RM(ASM)根据返回信息向 NM 申请资源。
NM 分配创建一个container 并创建Spark Application Master(AM）,此时 AM 上运行的是 Spark Driver。（每个SparkContext都有一个 AM）
AM启动后，和RM(ASM)通讯，请求根据任务信息向RM(ASM)申请 container 来启动 executor
RM(ASM)将申请到的资源信息返回给AM
AM 根据返回的资源信息区请求对应的 NM 分配 container 来启动 executor
NM 收到请求会启动相应的 container 并启动 executor
executor 启动成后反向向 AM 注册
executor 和 AM 交互完成任务
后续的DAGScheduler、TaskScheduler、Shuffle等操作都是和standaloe一样
等到所有的任务执行完毕后，AM 向 ASM 取消注册并释放资源

19. Spark join的优化经验

Spark作为分布式的计算框架，最为影响其执行效率的地方就是频繁的网络IO传输。所以，一般在不存在数据倾斜的情况下，想要提高Spark Job的执行效率，就尽量减少Job的Shuffle过程（减少Job的Stage），或者减少Shuffle带来的影响。

尽量减少参与join的RDD数据量
尽量避免参与join的RDD都具有重复的Key
尽量避免或者减少Shuffle过程
条件允许的情况下，使用map-join完成join

20. Spark的Shuffle

【Spark Core之Shuffle解析】【Spark Shuffle原理及相关调优】

Spark Shuffle演进的历史：

Spark 0.8及以前 Hash Based Shuffle
Spark 0.8.1 为Hash Based Shuffle引入File Consolidation机制【Hash Shuffle v2】
Spark 0.9 引入ExternalAppendOnlyMap
Spark 1.1 引入Sort Based Shuffle，但默认仍为Hash Based Shuffle
Spark 1.2 默认的Shuffle方式改为Sort Based Shuffle
Spark 1.4 引入Tungsten-Sort Based Shuffle【Unsafe Shuffle】
Spark 1.6 Tungsten-sort并入Sort Based Shuffle【Sort Shuffle V2】
Spark 2.0 Hash Based Shuffle退出历史舞台

配置spark.shuffle.manager
目前只有SortShuffleManager

HashShuffleManager

数据不进行排序，速度比较快
直接写入缓冲区，缓冲区写满后溢写为文件
ShuffleMapStage的每一个Task会生成与下一个ShuffleMapStage并行度相同的文件数量
海量文件操作句柄和临时缓存信息，占用内存容易内存溢出

SortShuffleManager

会对数据进行排序
在写入缓存之前，如果是reduceByKey之类的算子，则会先写入到一个Map内存数据结构中，而如果是join之类的算子，则先写入到Array内存数据结构中。在每条数据写入前先判断是否到达一定阈值，达到了就写入缓冲区。
复用一个Core的Task会写到同一个文件里面，并生成一个索引文件。其中记录了下一个ShuffleMapStage中每一个Task所要拉取数据的start offset和end offset。

21. 哪些算子操作涉及到Shuffle

distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition

22. MapReduce的Shuffle VS Spark的Shuffle

从整体功能上看，两者并没有大的差别。都是将 mapper（Spark 里是 ShuffleMapTask）的输出进行 partition，不同的 partition 送到不同的 reducer（Spark 里 reducer 可能是下一个 stage 里的 ShuffleMapTask，也可能是 ResultTask）。Reducer 以内存作缓冲区，边 shuffle 边 aggregate 数据，等到数据 aggregate 好以后进行 reduce（Spark 里可能是后续的一系列操作）。

从流程的上看，两者差别不小。Hadoop MapReduce 是 sort-based，进入 combine和 reduce的 records 必须先 sort。这样的好处在于 combine/reduce可以处理大规模的数据，因为其输入数据可以通过外排得到（mapper 对每段数据先做排序，reducer 的 shuffle 对排好序的每段数据做归并）。以前 Spark 默认选择的是 hash-based，通常使用 HashMap 来对 shuffle 来的数据进行合并，不会对数据进行提前排序。如果用户需要经过排序的数据，那么需要自己调用类似 sortByKey的操作。在Spark 1.2之后，sort-based变为默认的Shuffle实现。

从流程实现角度来看，两者也有不少差别。Hadoop MapReduce 将处理流程划分出明显的几个阶段：map, spill, merge, shuffle, sort, reduce等。每个阶段各司其职，可以按照过程式的编程思想来逐一实现每个阶段的功能。在 Spark 中，没有这样功能明确的阶段，只有不同的 stage 和一系列的 transformation，所以 spill, merge, aggregate 等操作需要蕴含在 transformation中。

与MapReduce完全不一样的是，在MapReduce中，map task必须将所有的数据都写入本地磁盘文件以后，才能启动reduce操作，来拉取数据。为什么？因为mapreduce要实现默认的根据key的排序！所以要排序，肯定得写完所有数据，才能排序，然后reduce来拉取。

但是Spark不需要，spark默认情况下，是不会对数据进行排序的。因此ShuffleMapTask每写入一点数据，ResultTask就可以拉取一点数据，然后在本地执行我们定义的聚合函数和算子，进行计算。
spark这种机制的好处在于，速度比mapreduce快多了。但是也有一个问题，mapreduce提供的reduce，是可以处理每个key对应的value上的，很方便。但是spark中，由于这种实时拉取的机制，因此提供不了直接处理key对应的values的算子，只能通过groupByKey，先shuffle，产生一个MapPartitionsRDD，然后用map算子，来处理每个key对应的values，就没有mapreduce的计算模型那么方便。

23. Spark广播变量的作用

使用广播变量，每个Executor的内存中，只驻留一份变量副本，而不是对每个Task都传输一次大变量，省了很多的网络传输，对性能提升具有很大帮助，而且会通过高效的广播算法来减少传输代价。
使用广播变量的场景：小表广播，使用map join代替reduce join，通过把小的数据集广播到各个节点上，节省了一次特别expense的Shuffle操作。
比如Driver上有一张数据量很小的表，其他节点上的Task都需要lookup这张表，那么Driver可以先把这张表copy到这些节点，这样Task就可以在本地查表了。

24. 数据倾斜

数据倾斜的发生，一般都是一个或者某几个Key对应的数据过大，导致Task执行过慢，或者内存溢出OOM，一般发生在Shuffle的时候，比如reduceByKey、countByKey、groupByKey，容易产生数据倾斜。

解决问题：首先看log日志信息，因为log日志报错的时候会提示在哪一行，然后去检查发生Shuffle的地方，这些地方比较容易发生数据倾斜：

聚合源数据
假设数据一般来源于Hive表，那么在生成Hive表的时候对数据进行聚合，按照Key进行分组，将Key对应的所有values以另一种格式存储，比如拼接成一个字符串，就不用Shuffle了，也就不会出现数据倾斜。
过滤导致倾斜的Key
与实际业务有关，把大的Key进行过滤掉
提高Shuffle操作Reduce的并行度
(reduceByKey(new..,1000) 通过提高Reduce端的Task执行数量，来分担数据压力，也就是说将Task执行数量提高，性能也会相应提高，这样方式如果在运行中确实解决了数据倾斜问题最好，但是如果出现数据倾斜之前OOM了，加大了Reduce端Task数量后可以运行了，但是执行时间变长，就放弃这种方案，看看背得。
双重聚合
用于groupByKey和reduceByKey，或者join。第一轮Key进行打散，将原来一样的Key变成不一样的Key前面加前缀，相当于将一样的Key分了多个组，然后进行局部聚合，接着除掉每个Key的前缀，然后再进行全局的聚合，进行两次聚合，避免数据倾斜问题。
将reduce join转换成map join 小表broadcast广播，每个节点的blockmanager都有一份。就不会发生Shuffle，不存在数据倾斜。或者大表加大内存。
Sample抽样分解聚合
使用随机数和扩容进行join 通过flatmap进行扩容，然后再将随机数打入进去进行join

25. RDD默认分区数

defaultParallelism与CPU的核数有关，默认是CPU的核数和与2进行比较的最小值。

26. 100个分片，聚合成两个分片

使用coalesce算子，主要就是用于在filter操作之后，针对每个Partition的数量各不相同的情况，来压缩Partition的数据。减少Partition的数量，而且让每个Partition的数据量都尽量均匀紧凑。从而方便后面的Task进行计算操作，能够在一定程度上提升性能。

27. Spark的通信机制

Spark消息通信主要分成三个部分：整体框架；启动消息通信；运行时消息通信。

早期Akka；

Actor并发模型，高可靠、高性能、可扩展

Spark2.x + Netty

Spark2.x版本使用Netty通讯框架作为内部通讯组件。Spark 基于Netty新的RPC框架借鉴了Akka的中的设计，也是基于Actor模型。

很多Spark用户也使用Akka，但是由于Akka不同版本之间无法互相通信，这就要求用户必须使用跟Spark完全一样的Akka版本，导致用户无法升级Akka。

Spark的Akka配置是针对Spark自身来调优的，可能跟用户自己代码中的Akka配置冲突。

Spark用的Akka特性很少，这部分特性很容易自己实现。同时，这部分代码量相比Akka来说少很多，debug比较容易。如果遇到什么bug，也可以自己马上fix，不需要等Akka上游发布新版本。而且，Spark升级Akka本身又因为第一点会强制要求用户升级他们使用的Akka，对于某些用户来说是不现实的。

RpcEndpoint：RPC端点，Spark针对于每个节点（Client/Master/Worker）都称之一个RPC端点，且都实现RpcEndpoint接口，内部根据不同端点的需求，设计不同的消息和不同的业务处理，如果需要发送（询问）则调用Dispatcher
RpcEnv：RPC上下文环境，每个RPC端点运行时依赖的上下文环境称之为RpcEnv
Dispatcher：消息分发器，针对于RPC端点需要发送消息或者从远程RPC接收到的消息，分发至对应的指令收件箱/发件箱。如果指令接收方是自己存入收件箱，如果指令接收方为非自身端点，则放入发件箱
Inbox：指令消息收件箱，一个本地端点对应一个收件箱，Dispatcher在每次向Inbox存入消息时，都将对应EndpointData加入内部待Receiver Queue中，另外Dispatcher创建时会启动一个单独线程进行轮询Receiver Queue，进行收件箱消息消费
OutBox：指令消息发件箱，一个远程端点对应一个发件箱，当消息放入Outbox后，紧接着将消息通过TransportClient发送出去。消息放入发件箱以及发送过程是在同一个线程中进行，这样做的主要原因是远程消息分为RpcOutboxMessage, OneWayOutboxMessage两种消息，而针对于需要应答的消息直接发送且需要得到结果进行处理
TransportClient：Netty通信客户端，根据OutBox消息的receiver信息，请求对应远程TransportServer
TransportServer：Netty通信服务端，一个RPC端点一个TransportServer,接受远程消息后调用Dispatcher分发消息至对应收发件箱