关于面试:面试系列五-之-项目涉及技术Spark

1）Local: 运行在一台机器上，通常是练手或者测试环境。
2）Standalone: 构建一个基于 Mster+Slaves 的资源调度集群，Spark 工作提交给 Master 运行。是 Spark 本身的一个调度零碎。
3）Yarn: Spark 客户端间接连贯 Yarn，不须要额定构建 Spark 集群。有 yarn-client 和 yarn-cluster 两种模式，次要区别在于：Driver 程序的运行节点。
4）Mesos：国内大环境比拟少用。

Shell 脚本。

1）在提交工作时的几个重要参数

executor-cores —— 每个 executor 应用的内核数，默认为 1，官网倡议 2 - 5 个，咱们企业是 4 个
num-executors —— 启动 executors 的数量，默认为 2
executor-memory —— executor 内存大小，默认 1G
driver-cores —— driver 应用内核数，默认为 1
driver-memory —— driver 内存大小，默认 512M

2）边给一个提交工作的款式

spark-submit \

  --master local\[5\]  \

  --driver-cores 2   \

  --driver-memory 8g \

  --executor-cores 4 \

  --num-executors 10 \

  --executor-memory 8g \

  --class PackageName.ClassName XXXX.jar \

  --name "Spark Job Name" \

  InputPath      \

  OutputPath

参考：https://blog.csdn.net/wuxintd…

RDD在 Lineage 依赖方面分为两种 Narrow Dependencies 与Wide Dependencies用来解决数据容错时的高效性以及划分工作时候起到重要作用。

Stage：依据 RDD 之间的依赖关系的不同将 Job 划分成不同的 Stage，遇到一个宽依赖则划分一个 Stage。

Task：Stage 是一个 TaskSet，将 Stage 依据分区数划分成一个个的 Task。

参考：https://blog.csdn.net/wuxintd…

1）map（func）：返回一个新的 RDD，该 RDD 由每一个输出元素通过 func 函数转换后组成.

2）mapPartitions(func)：相似于 map，但独立地在 RDD 的每一个分片上运行，因而在类型为 T 的 RD 上运行时，func 的函数类型必须是 Iterator[T] => Iterator[U]。假如有 N 个元素，有 M 个分区，那么 map 的函数的将被调用 N 次, 而 mapPartitions 被调用 M 次, 一个函数一次解决所有分区。

3）reduceByKey（func，[numTask]）：在一个 (K,V) 的 RDD 上调用，返回一个 (K,V) 的 RDD，应用定的 reduce 函数，将雷同 key 的值聚合到一起，reduce 工作的个数能够通过第二个可选的参数来设置。

4）aggregateByKey (zeroValue:U,[partitioner: Partitioner]) (seqOp: (U, V) => U,combOp: (U, U) => U: 在 kv 对的 RDD 中，，按 key 将 value 进行分组合并，合并时，将每个 value 和初始值作为 seq 函数的参数，进行计算，返回的后果作为一个新的 kv 对，而后再将后果依照 key 进行合并，最初将每个分组的 value 传递给 combine 函数进行计算（先将前两个 value 进行计算，将返回后果和下一个 value 传给 combine 函数，以此类推），将 key 与计算结果作为一个新的 kv 对输入。

5）combineByKey(createCombiner: V=>C, mergeValue: (C, V) =>C, mergeCombiners: (C, C) =>C):

对雷同 K，把 V 合并成一个汇合。

1.createCombiner: combineByKey() 会遍历分区中的所有元素，因而每个元素的键要么还没有遇到过，要么就和之前的某个元素的键雷同。如果这是一个新的元素,combineByKey()会应用一个叫作 createCombiner()的函数来创立那个键对应的累加器的初始值

2.mergeValue: 如果这是一个在解决以后分区之前曾经遇到的键，它会应用 mergeValue()办法将该键的累加器对应的以后值与这个新的值进行合并

3.mergeCombiners: 因为每个分区都是独立解决的，因而对于同一个键能够有多个累加器。如果有两个或者更多的分区都有对应同一个键的累加器，就须要应用用户提供的 mergeCombiners() 办法将各个分区的后果进行合并。

…

依据本身状况抉择比拟相熟的算子加以介绍。

参考：https://blog.csdn.net/wuxintd…

1）reduce：

2）collect:

3）first：

4）take：

5）aggregate：

6）countByKey：

7）foreach：

8）saveAsTextFile：

reduceBykey：

groupByKey：

…

ByKey:

未经优化的 HashShuffle：

优化后的 Shuffle：

一般的 SortShuffle：

当 shuffle read task 的数量小于等于 spark.shuffle.sort。

bypassMergeThreshold 参数的值时（默认为 200），就会 启用 bypass 机制。

reduceByKey：依照 key 进行聚合，在 shuffle 之前有 combine（预聚合）操作，返回后果是 RDD[k,v]。

groupByKey：依照 key 进行分组，间接进行 shuffle。

开发领导：reduceByKey 比 groupByKey，倡议应用。然而须要留神是否会影响业务逻辑。

1）关系：

两者都是用来扭转 RDD 的 partition 数量的，repartition 底层调用的就是 coalesce 办法：coalesce(numPartitions, shuffle = true)

2）区别：

repartition 肯定会产生 shuffle，coalesce 依据传入的参数来判断是否产生 shuffle

个别状况下

增大 rdd 的 partition 数量应用 repartition
缩小 partition 数量时应用 coalesce

都是做 RDD 长久化的

cache: 内存，不会截断血缘关系，应用计算过程中的数据缓存。

checkpoint：磁盘，截断血缘关系，在 ck 之前必须没有任何工作提交才会失效，ck 过程会额定提交一次工作。

累加器（accumulator）是 Spark 中提供的一种分布式的变量机制，其原理相似于 mapreduce，即分布式的扭转，而后聚合这些扭转。累加器的一个常见用处是在调试时对作业执行过程中的事件进行计数。而播送变量用来高效散发较大的对象。

共享变量呈现的起因：

通常在向 Spark 传递函数时，比方应用 map() 函数或者用 filter() 传条件时，能够应用驱动器程序中定义的变量，然而集群中运行的每个工作都会失去这些变量的一份新的正本，更新这些正本的值也不会影响驱动器中的对应变量。

Spark 的两个共享变量，累加器与播送变量，别离为后果聚合与播送这两种常见的通信模式冲破了这一限度。

应用 foreachPartition 代替 foreach，在 foreachPartition 内获取数据库的连贯。

1）RDD

长处:

编译时类型平安
编译时就能查看出类型谬误
面向对象的编程格调
间接通过类名点的形式来操作数据

毛病:

序列化和反序列化的性能开销
无论是集群间的通信, 还是 IO 操作都须要对对象的构造和数据进行序列化和反序列化。
GC 的性能开销，频繁的创立和销毁对象, 势必会减少 GC

2）DataFrame

DataFrame 引入了 schema 和off-heap

schema : RDD 每一行的数据, 构造都是一样的，这个构造就存储在 schema 中。Spark 通过 schema 就可能读懂数据, 因而在通信和 IO 时就只须要序列化和反序列化数据, 而构造的局部就能够省略了。

3）DataSet

DataSet 联合了 RDD 和 DataFrame 的长处，并带来的一个新的概念Encoder。

当序列化数据时，Encoder 产生字节码与 off-heap 进行交互，可能达到按需拜访数据的成果，而不必反序列化整个对象。Spark 还没有提供自定义 Encoder 的 API，然而将来会退出。

三者之间的转换：

join 和 sql 中的 inner join 操作很类似，返回后果是后面一个汇合和前面一个汇合中匹配胜利的，过滤掉关联不上的。

leftJoin 相似于 SQL 中的左外关联 left outer join，返回后果以第一个 RDD 为主，关联不上的记录为空。

局部场景下能够应用 left semi join 代替 left join：

因为 left semi join 是 in(keySet) 的关系，遇到右表重复记录，左表会跳过 , 性能更高 ，而 left join 则会始终遍历。 然而 left semi join 中最初 select 的后果中只许呈现左表中的列名，因为右表只有 join key 参加关联计算了


 val conf: SparkConf = new SparkConf().setMaster("local\[*\]").setAppName("WordCount")

 val sc = new SparkContext(conf)

 sc.textFile("/input")

.flatMap(_.split(" "))

.map((_,1))

.reduceByKey(_+_)

.saveAsTextFile("/output")

 sc.stop()

办法 1：

（1）依照 key 对数据进行聚合（groupByKey）
（2）将 value 转换为数组，利用 scala 的 sortBy 或者 sortWith 进行排序（mapValues）数据量太大，会 OOM。

办法 2：

（1）取出所有的 key

（2）对 key 进行迭代，每次取出一个 key 利用 spark 的排序算子进行排序

办法 3：

（1）自定义分区器，依照 key 进行分区，使不同的 key 进到不同的分区

（2）对每个分区使用 spark 的排序算子进行排序

这里举个例子。比方咱们有几百个文件，会有几百个 map 呈现，读取之后进行 join 操作，会十分的慢。这个时候咱们能够进行 coalesce 操作，比方 240 个 map，咱们合成 60 个 map，也就是窄依赖。这样再 shuffle，过程产生的文件数会大大减少。进步 join 的工夫性能。

参考：https://chbxw.blog.csdn.net/a…

1、基于 Receiver 的形式

这种形式应用 Receiver 来获取数据。Receiver 是应用 Kafka 的高层次 Consumer API 来实现的。receiver 从 Kafka 中获取的数据都是存储在 Spark Executor 的内存中的（如果忽然数据暴增，大量 batch 沉积，很容易呈现内存溢出的问题），而后 Spark Streaming 启动的 job 会去解决那些数据。

然而，在默认的配置下，这种形式可能会因为底层的失败而失落数据。如果要启用高牢靠机制，让数据零失落，就必须启用 Spark Streaming 的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接管到的 Kafka 数据写入分布式文件系统（比方 HDFS）上的预写日志中。所以，即便底层节点呈现了失败，也能够应用预写日志中的数据进行复原。

2、基于 Direct 的形式

这种新的不基于 Receiver 的间接形式，是在 Spark 1.3 中引入的，从而可能确保更加强壮的机制。代替掉应用 Receiver 来接收数据后，这种形式会周期性地查问 Kafka，来取得每个 topic+partition 的最新的 offset，从而定义每个 batch 的 offset 的范畴。当解决数据的 job 启动时，就会应用 Kafka 的简略 consumer api 来获取 Kafka 指定 offset 范畴的数据。

长处如下：

简化并行读取：如果要读取多个 partition，不须要创立多个输出 DStream 而后对它们进行 union 操作。Spark 会创立跟 Kafka partition 一样多的 RDD partition，并且会并行从 Kafka 中读取数据。所以在 Kafka partition 和 RDD partition 之间，有一个一对一的映射关系。

高性能：如果要保障零数据失落，在基于 receiver 的形式中，须要开启 WAL 机制。这种形式其实效率低下，因为数据实际上被复制了两份，Kafka 本人自身就有高牢靠的机制，会对数据复制一份，而这里又会复制一份到 WAL 中。而基于 direct 的形式，不依赖 Receiver，不须要开启 WAL 机制，只有 Kafka 中作了数据的复制，那么就能够通过 Kafka 的正本进行复原。

一次且仅一次的事务机制。

3、比照：

基于 receiver 的形式，是应用 Kafka 的高阶 API 来在 ZooKeeper 中保留生产过的 offset 的。这是生产 Kafka 数据的传统形式。这种形式配合着 WAL 机制能够保证数据零失落的高可靠性，然而却无奈保证数据被解决一次且仅一次，可能会解决两次。因为 Spark 和 ZooKeeper 之间可能是不同步的。

基于 direct 的形式，应用 kafka 的简略 api，Spark Streaming 本人就负责追踪生产的 offset，并保留在 checkpoint 中。Spark 本人肯定是同步的，因而能够保证数据是生产一次且仅生产一次。

在理论生产环境中大都用 Direct 形式

窗口函数就是在原来定义的 SparkStreaming 计算批次大小的根底上再次进行封装，每次计算多个批次的数据，同时还须要传递一个滑动步长的参数，用来设置当次计算工作实现之后下一次从什么中央开始计算。

图中 time1 就是 SparkStreaming 计算批次大小，虚线框以及实线大框就是窗口的大小，必须为批次的整数倍。虚线框到大实线框的间隔（相隔多少批次），就是滑动步长。

关于面试:面试系列五-之-项目涉及技术Spark

一、Spark

1.1 Spark 有几种部署形式？请别离简要阐述

1.2 Spark 工作应用什么进行提交，javaEE 界面还是脚本

1.3 Spark 提交作业参数（重点）

1.4 简述 Spark 的架构与作业提交流程（画图解说，注明各个局部的作用）（重点）

1.4.1、standlone

1.4.2、yarn-cluster

1.5 如何了解 Spark 中的血统概念（RDD）（口试重点）

1.6 简述 Spark 的宽窄依赖，以及 Spark 如何划分 stage，每个 stage 又依据什么决定 task 个数?（口试重点）

1.7 请列举 Spark 的 transformation 算子（不少于 8 个），并简述性能（重点）

1.8 请列举 Spark 的 action 算子（不少于 6 个），并简述性能（重点）

1.9 请列举会引起 Shuffle 过程的 Spark 算子，并简述性能。

1.10 简述 Spark 的两种外围 Shuffle（HashShuffle 与 SortShuffle）的工作流程（包含未优化的 HashShuffle、优化的 HashShuffle、一般的 SortShuffle 与 bypass 的 SortShuffle）（重点）

1.11 Spark 罕用算子 reduceByKey 与 groupByKey 的区别，哪一种更具劣势？（重点）

1.12 Repartition 和 Coalesce 关系与区别

1.13 别离简述 Spark 中的缓存机制（cache 和 persist）与 checkpoint 机制，并指出两者的区别与分割

1.14 简述 Spark 中共享变量（播送变量和累加器）的基本原理与用处。（重点）

1.15 当 Spark 波及到数据库的操作时，如何缩小 Spark 运行中的数据库连接数？

1.16 简述 SparkSQL 中 RDD、DataFrame、DataSet 三者的区别与分割?（口试重点）

1.17 SparkSQL 中 join 操作与 left join 操作的区别？

1.18 请手写出 wordcount 的 Spark 代码实现（Scala）（手写代码重点）

1.19、如何应用 Spark 实现 topN 的获取（形容思路或应用伪代码）（重点）

1.20 京东：调优之前与调优之后性能的具体比照（例如调整 map 个数，map 个数之前多少、之后多少，有什么晋升）

二、SparkStreaming

2.1、SparkStreaming 有哪几种形式生产 Kafka 中的数据，它们之间的区别是什么？

2.2 简述 SparkStreaming 窗口函数的原理（重点）

三、SparkSQL

关注我的公众号【宝哥大数据】，更多干货