大数据面试之Spark

Bertha 。 2023-02-17 15:54 27阅读 0赞

### 大数据面试之Spark ###

*  1.Spark
 *   *  1.0 Spark架构
     *  1.1 Spark的Shuffle过程？与Hadoop的Shuffle过程对比
     *  1.2 Spark中reduceBykey和groupBykey的区别
     *  1.3 Spark中和repartition相似的算子？优缺点
     *  1.4 Spark的调优
     *  1.5 Spark中数据倾斜处理
     *  1.6 Spark的多种提交方式？python提交方式？
     *  1.7 Spark广播变量和累加器的使用
     *  1.8 Spark 3.X 新特性

说明，感谢亮哥长期对我的帮助，此处多篇文章均为亮哥带我整理。以及参考诸多博主的文章。如果侵权，请及时指出，我会立马停止该行为；如有不足之处，还请大佬不吝指教，以期共同进步。

# 1.Spark #

## 1.0 Spark架构 ##

Spark架构使用了分布式计算中master-slave模型，master是集群中含有master进程的节点，slave是集群中含有worker进程的节点。
    
    Driver Program ：运⾏main函数并且新建SparkContext的程序。
    
    Application：基于Spark的应用程序，包含了driver程序和集群上的executor。
    
    Cluster Manager：指的是在集群上获取资源的外部服务。目前有三种类型
        （1）Standalone: spark原生的资源管理，由Master负责资源的分配
        （2）Apache Mesos:与hadoop MR兼容性良好的一种资源调度框架
        （3）Hadoop Yarn: 主要是指Yarn中的ResourceManager
    
    Worker Node： 集群中任何可以运行Application代码的节点，在Standalone模式中指的是通过slaves文件配置的Worker节点，在Spark on Yarn模式下就是NodeManager节点
    
    Executor：是在一个worker node上为某应⽤启动的⼀个进程，该进程负责运⾏行任务，并且负责将数据存在内存或者磁盘上。每个应⽤都有各自独立的executor。
    
    Task ：被送到某个executor上的工作单元。

Spark角色作用

Driver（驱动器）
    
        Spark的驱动器是执行开发程序中的main方法的进程。它负责开发人员编写的用来
            创建SparkContext、创建RDD，以及进行RDD的转化操作和行动操作代码的执行
        
        1. 把用户程序转为作业（JOB）
        2. 跟踪Executor的运行状况
        3. 为执行器节点调度任务
        4. UI展示应用运行状况
    
    
    Executor（执行器）
    
        Spark Executor是一个工作进程，负责在 Spark 作业中运行任务，任务间相互独立。
            Spark 应用启动时，Executor节点被同时启动，并且始终伴随着整个 Spark 应用的生命周期而存在。
            如果有Executor节点发生了故障或崩溃，Spark 应用也可以继续执行，
            会将出错节点上的任务调度到其他Executor节点上继续运行
    
        1. 负责运行组成 Spark 应用的任务，并将结果返回给驱动器进程
        2. 通过自身的块管理器（Block Manager）为用户程序中要求缓存的RDD提供内存式存储。RDD是直接缓存在Executor进程内的，因此任务可以在运行时充分利用缓存数据加速运算
    
    Master
        是Standalone资源调度框架里面资源管理的主节点。也是JVM进程。

[参考][Link 1]

spark 常见端口及功能

namenode的webUI端口：50070
    yarn的web端口：8088
    spark集群的web端口：8080
    spark-job监控端口：4040
    job history 端口:18080

[参考][Link 2]

## 1.1 Spark的Shuffle过程？与Hadoop的Shuffle过程对比 ##

Hadoop：map端保存分片数据，通过网络收集到reduce端。
    
    Spark：spark的shuffle实在DAGSchedular划分Stage的时候产生的，TaskSchedular要分发Stage到各个worker的executor。减少shuffle可以提高性能。
    
    1.	SparkShuffle概念
        reduceByKey会将上一个RDD中的每一个key对应的所有value聚合成一个value，然后生成一个新的RDD，元素类型是<key,value>对的形式，这样每一个key对应一个聚合起来的value。
    
    问题：聚合之前，每一个key对应的value不一定都是在一个partition中，也不太可能在同一个节点上，因为RDD是分布式的弹性的数据集，RDD的partition极有可能分布在各个节点上。
    
    如何聚合？
    – Shuffle Write：上一个stage的每个map task就必须保证将自己处理的当前分区的数据相同的key写入一个分区文件中，可能会写入多个不同的分区文件中。
    
     – Shuffle Read：reduce task就会从上一个stage的所有task所在的机器上寻找属于己的那些分区文件，这样就可以保证每一个key所对应的value都会汇聚到同一个节点上去处理和聚合。
     
    Spark中有两种Shuffle类型，HashShuffle和SortShuffle，Spark1.2之前是HashShuffle默认的分区器是HashPartitioner，Spark1.2引入SortShuffle默认的分区器是RangePartitioner。

[参考][Link 3]  
[参考][Link 4]

## 1.2 Spark中reduceBykey和groupBykey的区别 ##

相同点：
        1. 都作用于 RDD[K,V]
        2. 都是根据key来分组聚合
        3. 默认，分区的数量都是不变的，但是都可以通过参数来指定分区数量
    
    不同点：
        1. groupByKey默认没有聚合函数，得到的返回值类型是RDD[ k,Iterable[V]]
        2. reduceByKey 必须传聚合函数 得到的返回值类型 RDD[(K,聚合后的V)]
        3. groupByKey().map() = reduceByKey
    
    最重要的区别：
        reduceByKey 会进行分区内聚合，然后再进行网络传输
        groupByKey 不会进行局部聚合
    
    结论：
        如果这两个算子，都可以使用， 优先使用reduceByKey

## 1.3 Spark中和repartition相似的算子？优缺点 ##

repartition(numPartitions) 重新分区 必须shuffle  参数是要分多少区  少变多
    repartitionAndSortWithinPartitions(partitioner) 重新分区+排序  比先分区再排序效率高  对K/V的RDD进行操作
    
    repartitionAndSortWithinPartitions根据给定的分区程序对RDD进行重新分区，并在每个生成的分区内按键对记录进行排序。
    这比调用重新分区，然后在每个分区内进行排序更有效率，因为它可以将排序压入洗牌机器。
    
    什么时候使用repartitionAndSortWithinPartitions？
        如果需要重分区，并且想要对分区中的数据进行升序排序。
        提高性能，替换repartition和sortBy
    
    val array=Array(2,4,6,67,3,45,26,35,789,345)
      val data=sc.parallelize(array)
      // 替换repartition组合sortBy
      data.zipWithIndex().repartitionAndSortWithinPartitions(new HashPartitioner(1)).foreach(println)

[参考][Link 5]

[参考2][2]

## 1.4 Spark的调优 ##

1. 避免创建重复的RDD，复用同一个RDD
    
    2. 对多次使用的RDD进行持久化
        持久化算子：
        cache:
        	MEMORY_ONLY
        persist：
        	MEMORY_ONLY
        	MEMORY_ONLY_SER
        	MEMORY_AND_DISK_SER
        	一般不要选择带有_2的持久化级别。
        checkpoint:
        ① 如果一个RDD的计算时间比较长或者计算起来比较复杂，一般将这个RDD的计算结果保存到HDFS上，这样数据会更加安全。
        ② 如果一个RDD的依赖关系非常长，也会使用checkpoint,会切断依赖关系，提高容错的效率。
        
    3. 尽量避免使用shuffle类的算子
        使用广播变量来模拟使用join,使用情况：一个RDD比较大，一个RDD比较小。
        join算子=广播变量+filter、广播变量+map、广播变量+flatMap
        
    4. 使用map-side预聚合的shuffle操作
        即尽量使用有combiner的shuffle类算子。
        combiner概念：
        	在map端，每一个map task计算完毕后进行的局部聚合。
        combiner好处：
            1)	降低shuffle write写磁盘的数据量。
            2)	降低shuffle read拉取数据量的大小。
            3)	降低reduce端聚合的次数。
        有combiner的shuffle类算子：
        1)	reduceByKey:这个算子在map端是有combiner的，在一些场景中可以使用reduceByKey代替groupByKey。
        2)	aggregateByKey
        3)	combineByKey
        
    5. 尽量使用高性能的算子
        使用reduceByKey替代groupByKey
        使用mapPartition替代map
        使用foreachPartition替代foreach
        filter后使用coalesce减少分区数
        使用使用repartitionAndSortWithinPartitions替代repartition与sort类操作
        使用repartition和coalesce算子操作分区。
    
    6.	使用广播变量
        广播后的变量，会保证每个Executor的内存中，只驻留一份变量副本，而Executor中的task执行时共享该Executor中的那份变量副本。这样的话，可以大大减少变量副本的数量，从而减少网络传输的性能开销，并减少对Executor内存的占用开销，降低GC的频率。
    
    7. 使用Kryo优化序列化性能
        在Spark中，主要有三个地方涉及到了序列化：
        1)	在算子函数中使用到外部变量时，该变量会被序列化后进行网络传输。
        2)	将自定义的类型作为RDD的泛型类型时（比如JavaRDD<SXT>，SXT是自定义类型），所有自定义类型对象，都会进行序列化。因此这种情况下，也要求自定义的类必须实现Serializable接口。
        3)	使用可序列化的持久化策略时（比如MEMORY_ONLY_SER），Spark会将RDD中的每个partition都序列化成一个大的字节数组。
    
        Spark中使用Kryo：
            Sparkconf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
            .registerKryoClasses(new Class[]{SpeedSortKey.class})
    
    8.	优化数据结构
        java中有三种类型比较消耗内存：
        1) 对象，每个Java对象都有对象头、引用等额外的信息，因此比较占用内存空间。
        2) 字符串，每个字符串内部都有一个字符数组以及长度等额外信息。
        3) 集合类型，比如HashMap、LinkedList等，因为集合类型内部通常会使用一些内部类来封装集合元素，比如Map.Entry。
        
        因此Spark官方建议，在Spark编码实现中，特别是对于算子函数中的代码，尽量不要使用上述三种数据结构，尽量使用字符串替代对象，使用原始类型（比如Int、Long）替代字符串，使用数组替代集合类型，这样尽可能地减少内存占用，从而降低GC频率，提升性能。
    
    9.使用高性能的库fastutil

## 1.5 Spark中数据倾斜处理 ##

1.  使用Hive ETL预处理数据

方案适用场景：
        如果导致数据倾斜的是Hive表。如果该Hive表中的数据本身很不均匀（比如某个key对应了100万数据，其他key才对应了10条数据），而且业务场景需要频繁使用Spark对Hive表执行某个分析操作，那么比较适合使用这种技术方案。
    方案实现思路：
        此时可以评估一下，是否可以通过Hive来进行数据预处理（即通过Hive ETL预先对数据按照key进行聚合，或者是预先和其他表进行join），然后在Spark作业中针对的数据源就不是原来的Hive表了，而是预处理后的Hive表。此时由于数据已经预先进行过聚合或join操作了，那么在Spark作业中也就不需要使用原先的shuffle类算子执行这类操作了。
    方案实现原理：
        这种方案从根源上解决了数据倾斜，因为彻底避免了在Spark中执行shuffle类算子，那么肯定就不会有数据倾斜的问题了。但是这里也要提醒一下大家，这种方式属于治标不治本。因为毕竟数据本身就存在分布不均匀的问题，所以Hive ETL中进行group by或者join等shuffle操作时，还是会出现数据倾斜，导致Hive ETL的速度很慢。我们只是把数据倾斜的发生提前到了Hive ETL中，避免Spark程序发生数据倾斜而已。

1.  过滤少数导致倾斜的key

方案适用场景：
        如果发现导致倾斜的key就少数几个，而且对计算本身的影响并不大的话，那么很适合使用这种方案。比如99%的key就对应10条数据，但是只有一个key对应了100万数据，从而导致了数据倾斜。
    方案实现思路：
        如果我们判断那少数几个数据量特别多的key，对作业的执行和计算结果不是特别重要的话，那么干脆就直接过滤掉那少数几个key。比如，在Spark SQL中可以使用where子句过滤掉这些key或者在Spark Core中对RDD执行filter算子过滤掉这些key。如果需要每次作业执行时，动态判定哪些key的数据量最多然后再进行过滤，那么可以使用sample算子对RDD进行采样，然后计算出每个key的数量，取数据量最多的key过滤掉即可。
    方案实现原理：
        将导致数据倾斜的key给过滤掉之后，这些key就不会参与计算了，自然不可能产生数据倾斜。

1.  提高shuffle操作的并行度

方案实现思路：
        在对RDD执行shuffle算子时，给shuffle算子传入一个参数，比如reduceByKey(1000)，该参数就设置了这个shuffle算子执行时shuffle read task的数量。对于Spark SQL中的shuffle类语句，比如group by、join等，需要设置一个参数，即spark.sql.shuffle.partitions，该参数代表了shuffle read task的并行度，该值默认是200，对于很多场景来说都有点过小。
    方案实现原理：
        增加shuffle read task的数量，可以让原本分配给一个task的多个key分配给多个task，从而让每个task处理比原来更少的数据。举例来说，如果原本有5个不同的key，每个key对应10条数据，这5个key都是分配给一个task的，那么这个task就要处理50条数据。而增加了shuffle read task以后，每个task就分配到一个key，即每个task就处理10条数据，那么自然每个task的执行时间都会变短了。

1.  双重聚合

方案适用场景：
        对RDD执行reduceByKey等聚合类shuffle算子或者在Spark SQL中使用group by语句进行分组聚合时，比较适用这种方案。
    方案实现思路：
        这个方案的核心实现思路就是进行两阶段聚合。第一次是局部聚合，先给每个key都打上一个随机数，比如10以内的随机数，此时原先一样的key就变成不一样的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)，就会变成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)。接着对打上随机数后的数据，执行reduceByKey等聚合操作，进行局部聚合，那么局部聚合结果，就会变成了(1_hello, 2) (2_hello, 2)。然后将各个key的前缀给去掉，就会变成(hello,2)(hello,2)，再次进行全局聚合操作，就可以得到最终结果了，比如(hello, 4)。
    方案实现原理：
        将原本相同的key通过附加随机前缀的方式，变成多个不同的key，就可以让原本被一个task处理的数据分散到多个task上去做局部聚合，进而解决单个task处理数据量过多的问题。接着去除掉随机前缀，再次进行全局聚合，就可以得到最终的结果。

1.  将reduce join转为map join

BroadCast+filter(或者map)
    方案适用场景：
        在对RDD使用join类操作，或者是在Spark SQL中使用join语句时，而且join操作中的一个RDD或表的数据量比较小（比如几百M或者一两G），比较适用此方案。
    方案实现思路：
        不使用join算子进行连接操作，而使用Broadcast变量与map类算子实现join操作，进而完全规避掉shuffle类的操作，彻底避免数据倾斜的发生和出现。将较小RDD中的数据直接通过collect算子拉取到Driver端的内存中来，然后对其创建一个Broadcast变量；接着对另外一个RDD执行map类算子，在算子函数内，从Broadcast变量中获取较小RDD的全量数据，与当前RDD的每一条数据按照连接key进行比对，如果连接key相同的话，那么就将两个RDD的数据用你需要的方式连接起来。
    方案实现原理：
        普通的join是会走shuffle过程的，而一旦shuffle，就相当于会将相同key的数据拉取到一个shuffle read task中再进行join，此时就是reduce join。但是如果一个RDD是比较小的，则可以采用广播小RDD全量数据+map算子来实现与join同样的效果，也就是map join，此时就不会发生shuffle操作，也就不会发生数据倾斜。

1.  采样倾斜key并分拆join操作

方案适用场景：
        两个RDD/Hive表进行join的时候，如果数据量都比较大，无法采用“解决方案五”，那么此时可以看一下两个RDD/Hive表中的key分布情况。如果出现数据倾斜，是因为其中某一个RDD/Hive表中的少数几个key的数据量过大，而另一个RDD/Hive表中的所有key都分布比较均匀，那么采用这个解决方案是比较合适的。
    方案实现思路：
        对包含少数几个数据量过大的key的那个RDD，通过sample算子采样出一份样本来，然后统计一下每个key的数量，计算出来数据量最大的是哪几个key。然后将这几个key对应的数据从原来的RDD中拆分出来，形成一个单独的RDD，并给每个key都打上n以内的随机数作为前缀，而不会导致倾斜的大部分key形成另外一个RDD。接着将需要join的另一个RDD，也过滤出来那几个倾斜key对应的数据并形成一个单独的RDD，将每条数据膨胀成n条数据，这n条数据都按顺序附加一个0~n的前缀，不会导致倾斜的大部分key也形成另外一个RDD。再将附加了随机前缀的独立RDD与另一个膨胀n倍的独立RDD进行join，此时就可以将原先相同的key打散成n份，分散到多个task中去进行join了。而另外两个普通的RDD就照常join即可。最后将两次join的结果使用union算子合并起来即可，就是最终的join结果	。

1.  使用随机前缀和扩容RDD进行join

方案适用场景：
        如果在进行join操作时，RDD中有大量的key导致数据倾斜，那么进行分拆key也没什么意义，此时就只能使用最后一种方案来解决问题了。
    方案实现思路：
        该方案的实现思路基本和“解决方案六”类似，首先查看RDD/Hive表中的数据分布情况，找到那个造成数据倾斜的RDD/Hive表，比如有多个key都对应了超过1万条数据。然后将该RDD的每条数据都打上一个n以内的随机前缀。同时对另外一个正常的RDD进行扩容，将每条数据都扩容成n条数据，扩容出来的每条数据都依次打上一个0~n的前缀。最后将两个处理后的RDD进行join即可。

[参考][Link 6]

## 1.6 Spark的多种提交方式？python提交方式？ ##

Spark运行模式

Local
        多用于本地测试，如在eclipse，idea中写程序测试等。
    Standalone
        Standalone是Spark自带的一个资源调度框架，它支持完全分布式。
    Yarn
        Hadoop生态圈里面的一个资源调度框架，Spark也是可以基于Yarn来计算的。
    Mesos
    
    资源调度框架。
    
    要基于Yarn来进行资源调度，必须实现AppalicationMaster接口，Spark实现了这个接口，所以可以基于Yarn。

1.  Standalone-client提交任务方式

./spark-submit 
    --master spark://node1:7077 
    --class org.apache.spark.examples.SparkPi 
    ../lib/spark-examples-1.6.0-hadoop2.6.0.jar 
    1000
    
    或
    ./spark-submit 
    --master spark://node1:7077 
    --deploy-mode client 
    --class org.apache.spark.examples.SparkPi 
    ../lib/spark-examples-1.6.0-hadoop2.6.0.jar 
    100

1.  Standalone-cluster提交任务方式

./spark-submit 
    --master spark://node1:7077 
    --deploy-mode cluster
    --class org.apache.spark.examples.SparkPi 
    ../lib/spark-examples-1.6.0-hadoop2.6.0.jar 
    100

1.  yarn-client提交任务方式

./spark-submit 
    --master yarn
     --class org.apache.spark.examples.SparkPi ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
    100
    
    或
    ./spark-submit 
    --master yarn–client
     --class org.apache.spark.examples.SparkPi ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
    100
    
    或
    ./spark-submit 
    --master yarn 
    --deploy-mode  client 
     --class org.apache.spark.examples.SparkPi ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
    100

1.  yarn-cluster提交任务方式

./spark-submit 
    --master yarn 
    --deploy-mode cluster 
    --class org.apache.spark.examples.SparkPi ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
    100
    
    或
    ./spark-submit 
    --master yarn-cluster
    --class org.apache.spark.examples.SparkPi ../lib/spark-examples-1.6.0-hadoop2.6.0.jar
    100

提交python脚本

spark-submit \
    --master local[2] \
    --num-executors 2 \
    --executor-memory 1G \
    --py-files /home/hadoop/Download/test/firstApp.py

## 1.7 Spark广播变量和累加器的使用 ##

1.  Spark广播变量

如何定义一个广播变量
        val a = 3
        val broadcast = sc.broadcast(a)
    
    如何还原一个广播变量
        val c = broadcast.value
    
    广播变量使用
        val conf = new SparkConf()
        conf.setMaster("local").setAppName("brocast")
        val sc = new SparkContext(conf)
        val list = List("hello hadoop")
        val broadCast = sc.broadcast(list)
        val lineRDD = sc.textFile("./words.txt")
        lineRDD.filter { x => broadCast.value.contains(x) }.foreach { println}
        sc.stop()
    
    定义广播变量注意点 
        变量一旦被定义为一个广播变量，那么这个变量只能读，不能修改
    
    注意事项
        1、能不能将一个RDD使用广播变量广播出去？
        
            不能，因为RDD是不存储数据的。可以将RDD的结果广播出去。
        
        2、 广播变量只能在Driver端定义，不能在Executor端定义。
        
        3、 在Driver端可以修改广播变量的值，在Executor端无法修改广播变量的值。
        
        4、如果executor端用到了Driver的变量，如果不使用广播变量在Executor有多少task就有多少Driver端的变量副本。
        
        5、如果Executor端用到了Driver的变量，如果使用广播变量在每个Executor中只有一份Driver端的变量副本。

1.  累加器

如何定义一个累加器 
        val a = sc.accumulator(0)
    
    如何还原一个累加器 
        val b = a.value
    
    累加器的使用
        val conf = new SparkConf()
        conf.setMaster("local").setAppName("accumulator")
        val sc = new SparkContext(conf)
        val accumulator = sc.accumulator(0)
        sc.textFile("./words.txt").foreach { x =>{accumulator.add(1)}}
        println(accumulator.value)
        sc.stop()
    
    注意事项
        1、 累加器在Driver端定义赋初始值，累加器只能在Driver端读取最后的值，在Excutor端更新。
        2、累加器不是一个调优的操作，因为如果不这样做，结果是错的

[参考][Link 5]

## 1.8 Spark 3.X 新特性 ##

1. 动态分区修剪（Dynamic Partition Pruning）
    2. 自适应查询执行（Adaptive Query Execution）
    3. 加速器感知调度（Accelerator-aware Scheduling）
    4. Apache Spark 3.0 将内置支持 GPU 调度
    5. Apache Spark DataSource V2
    6. 更好的 ANSI SQL 兼容
    7. SparkR 向量化读写

[参考][Link 7]

[Link 1]: https://www.cnblogs.com/xiangyuguan/p/11429724.html
[Link 2]: https://www.jianshu.com/p/808ad3ae7fde
[Link 3]: https://www.cnblogs.com/itboys/p/9226479.html
[Link 4]: https://blog.csdn.net/u012137473/article/details/85238150
[Link 5]: https://www.cnblogs.com/frankdeng/p/9301653.html
[2]: https://blog.csdn.net/high2011/article/details/80764275/
[Link 6]: https://www.cnblogs.com/qingyunzong/p/8946679.html
[Link 7]: https://www.cnblogs.com/w397090770/p/11875829.html