面试（涉及技术二）

六、Flume

6.1 Flume的组成

6.1.1 taildir source

　　1）断点续传

　　2）Apache1.7以及CDH1.6产生

　　3）若遇到无断点续传功能的source怎么办？

　　　　自定义

　　4）taildir挂了怎么办？

　　　　不会丢失数据，因为有断点续传，可能会有重复数据

　　5）如何处理重复数据？

　　　　（1）不处理：生产环境通常不处理，因为会影响传输效率

　　　　（2）处理：可以在taildir source中增加自定义事务；也可以在Hive、SparkStream、Flink中进行处理，分组去重或开窗取第一条、Redis的Set集合都能去重

　　6）taildir source是否支持递归遍历文件夹来读取文件？

　　　　不支持，但是可以自定义，递归遍历文件夹 + 读取文件

6.1.2 channel

　　1）file channel

　　　　数据存储于磁盘，可以通过配置dataDirs指向多个路径，每个路径对应不同的硬盘，增加Flume的吞吐量，可靠性高，但传输速度低，默认容量100万Event

　　2）memory channel

　　　　数据存储于内存，传输速度快，但可靠性差，默认100个Event

　　3）Kafka channel

　　　　数据存储于Kafka，基于磁盘，可靠性高，传输速度大于memory channel + kafka sink（原因是省去了sink阶段）

　　4）kafka channel哪个版本产生的？

　　　　Flume1.6产生，但当时还没火，还存在一些Bug，增加了数据额外的清洗工作，Flume1.7开始解决了Bug，然后就火了

　　5）生产环境如何选择？

　　　　（1）若下一级是Kafka，选kafka channel

　　　　（2）若对数据要求非常准确的公司，选file channel

　　　　（3）若是普通日志，可以选择memory channel

6.1.3 HDFS sink

6.1.4 事务

　　source到channel是put事务，channel到sink是take事务

6.2 Flume拦截器

6.2.1 拦截器

　　1）ETL拦截器：轻度清洗，过滤出Json格式不完整的数据

　　2）时间拦截器：提取日志时间作为分区的依据，避免零点漂移问题

6.2.2 自定义拦截器步骤

　　1）实现Interceptor

　　2）重写方法

　　　　（1）initialize初始化方法

　　　　（2）public Event intercept(Event event) 处理单个Event

　　　　（3）public List<Event> intercept(List<Event> events) 处理多个Event，在这个方法中调用Event intercept(Event event)

　　　　（4）close方法

　　3）静态内部类，实现Interceptor.Builder

　　4）打包，上传至flume/lib目录下

　　5）在配置文件中关联：全类名 + $builder

6.2.3 拦截器可以不用吗？

　　可以不用，在hive的dwd层或sparkStream中处理即可，也可以用，但会影响性能，不推荐用在实时性高的场景

6.3 Flume Channel选择器

　　1）复制（默认）：将全部source中的Event数据发往所有的channel，我们公司使用默认

　　2）多路复用：将全部source中的Event数据有选择的发往相应的channel

6.4 Flume监控器

　　1）采用ganglia监控，若监控到Flume尝试提交的次数远远大于最终成功的次数，说明Flume运行比较差

　　2）解决办法

　　　　（1）增加内存：在flume-env.sh中设置内存大小为4~6G，-Xmx与-Xms最好设置一致，减少内存抖动带来的性能影响，不然容易导致频繁fullgc

　　　　（2）增加服务器数量：日志服务器配置8~16G内存、8T硬盘

6.5 Flume采集数据会丢失吗？

　　file channel 不会丢失数据，使用file channel 数据可以存储在磁盘中，并且数据传输自身有事务，memory channel可能会丢失数据或channel存储的数据已满，导致未写入的数据丢失

七、Kafka

7.1 Kafka架构

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

7.2 Kafka的机器数量

　　kafka机器数 = 2 * （峰值生产速度 * 副本数 / 100）+ 1，峰值生产速度可以通过生产者的压力测试得出，若峰值生产速度为50M/s，副本数为2，则Kafka机器数 = 2 * （50 * 2 / 100）+ 1 = 3台

7.3 副本数设定

　　我们公司设置为2个，提高可靠性，但增加了网络IO的传输

7.4 Kafka压测

　　kafka官方自带压力测试脚本，基本上每次此时都是网络IO率先达到瓶颈

7.5 Kafka日志保存时间

　　默认保存7天，我们公司保存3天

7.6 Kafka每天有多少数据量？

　　每天大概有60G的数据量，产生6千万条日志数据

　　平均每秒产生 60000000/24/60/60 = 700多条数据

　　低谷期每秒产生 20多条数据，也就是凌晨3点到5点这段时间

　　高峰期每秒产生 1万多条数据，也就是晚上7点到12点这段时间

　　每条日志数据平均大小为1k左右，所以平均每秒大约产生1M左右的数据

7.7 Kafka的硬盘大小

　　60G（每天产生的数据量） * 2（副本数）* 3（天）/  70%（预留30%的磁盘）= 500多G，我们公司设置为1T，完全够Kafka使用了

7.8 Kafka监控

　　我们公司使用的是开源组件：kafkaEagle

7.9 Kafka分区数

　　创建只有一个分区的主题，测试这个主题的生产者和消费者的吞吐量，假设他们的值分别是Tp和Tc，单位可以是MB/s，假设总的目标吞吐量是Tt，则分区数= Tt / min(Tp , Tc)

　　例如：producer吞吐量=20m/s；consumer吞吐量=50m/s，期望吞吐量100m/s；分区数=100 / 20 =5分区，分区数一般设置为：3-10个

　　我们公司设置为3个

7.10 Kafka的Topic数量

　　离线数仓中只有一个Topic，实时项目是从mysql中导入数据，所以有多少张表就有多少个Topic

7.11 Kafka的ISR副本同步队列

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

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

7.12 Kafka的分区分配策略

　　1）Range（默认）：Range是对每个Topic而言的，首先GroupCoordinator组协调器对同一个Topic里的分区按照序号排序，并对消费者按照字母顺序排序，然后用分区的个数除以消费者线程总数求出每个消费者线程消费的分区数，若除不尽，则前几个消费者线程多消费一个分区。这种方式在某种情况下可能导致数据倾斜

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

　　3）Sticky： 粘性的， 发生rebalance之后，尽量复用之前的分配关系

7.13 Kafka挂掉怎么办？

　　1）先重启试试，然后查看日志

　　2）数据可能会丢，当ack设置为1，leader收到数据后，未来得及同步数据就挂掉，这时可能会丢失数据

　　3）数据也可能重复，kafka没有事务，ack也没有设置为-1时，数据可能重复

　　4）我们公司的日志服务器有30天的备份，所以可以重新拉取一次数据即可

7.14 Kafka丢不丢数据？

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

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

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

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

7.15 Kafka数据重复问题

　　我们公司在kafka中加入了事务和幂等性，并且ack设置为-1，以此来保证数据不重复。当然，如果数据真的重复了，也可以在下游对数据进行处理，比如Hive的DWD层去重（分组或开窗取第一个值）、SparkStreaming、Flink都能去重，还有Redis的Set集合

7.16 Kafka数据积压怎么办？

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

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

7.17 Kafka的参数优化

　　1）日志保存策略有默认的7天改为3天

　　2）副本数由默认的1个改为2个

　　3）增大网络延时时间，如果网络不太好的话

　　4）生产者生产数据默认不进行压缩，我们可以采用snappy或lzo压缩，减少网络IO

　　5）每个Kafka服务的默认内存是1个G，可以改为5个G

7.18 Kafka为什么能高效的读写数据？

　　1）分布式集群，采用分区技术

　　2）顺序写磁盘

　　3）零拷贝

7.19 Kafka单条日志传输大小

　　默认是1M，我们公司的单条日志数据不会超过1M，所以使用默认值

7.20 Kafka过期数据的清洗

　　有两个方式进行处理：一种是压缩，一种是删除。我们公司是直接删除过期的数据

7.21 Kafka可以按照时间消费数据吗？

　　可以，有一个参数可以在消费者配置中进行设置，那个参数具体叫啥来着我记不清了

7.22 Kafka消费者角度考虑是拉取数据还是推送数据？

　　Kafka的消费策略我们可以称为事件驱动型，是一种被动型的消费，有数据过来我就消费，没有就闲着，像Flink也是一样，来一条数据就处理一条，不来就处于空闲状态

7.23 Kafka中的数据是有序的吗？

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

八、HBase

8.1 HBase的存储结构

8.2 HBase的RowKey设计原则

　　三大原则：长度、散列、唯一 三大原则

8.3 实际生产中如何设计RowKey?

　　1）生产随机数、散列值

　　2）字符串反转

8.4 Phoenix二级索引

8.4.1 Phoenix简介

　　1）定义：一款附着在Hbase上的皮肤

　　2）特点：

　　　　（1）易于集成，如Spark、Hive、MR、Flume

　　　　（2）操作简单，通过Phoenix操作HBase可以使用SQL化的语法进行操作

　　　　（3）支持HBase二级索引的创建

　　3）Phoenix架构

8.4.3 全局二级索引

　　使用Phoenix创建，CREATE INDEX teacher_index ON teacher(name);

8.4.4 本地二级索引

　　直接创建即可

　　面试官：怎么直接创建，在哪创建？

　　我们：创建表时创建，选择表中的某个字段，比如 CREATE LOCAL INDEX teacher_local_index ON teacher(name)

8.4.5 HBase协处理器（扩展）

　　1）案例需求：编写协处理器，实现在往A表插入数据的同时让HBase自身（协处理器）向B表中插入一条数据

　　2）实现步骤：

　　　　（1）在Shell窗口创建A表和B表

　　　　（2）创建Maven工程，引入依赖

　　　　（3）定义MyCoprocessor类并实现RegionObserver和RegionCoprocessor接口

　　　　（4）接下来可以通过重启的或不重启的方式加载该处理器

　　　　重启的方式：

　　　　（1）打包放入HBase的lib目录下

　　　　（2）分发jar包并重启HBase

　　　　（3）建表时指定注册协处理器即可

　　　　不重启的方式：

　　　　（1） 给hbase-site.xml中添加配置，防止协处理器异常导致集群停机

　　　　（2） 打成jar包并修改jar包名称为c1.jar然后至上传hdfs,比如存放路径为 hdfs://hadoop102:9820/c1.jar

　　　　（3） 在hbase的shell窗口禁用表

　　　　（4） 加载协处理器:alter '表名', METHOD => 'table_att', 'Coprocessor'=>'jar所处HDFS路径| 协处理器所在的全类名|优先级|参数'

　　　　（5） 启动表

　　　　（6) 向A表插入数据,观察B表是否同步有数据插入

九、Azkaban

9.1 每天集群多少指标？什么时候跑，跑多久？

　　每天大约跑100多个指标，有时会达到200多个左右；每天00:30开始跑，平均跑3个多小时

9.2 任务挂了怎么办？

　　1）运行成功或者失败都会发邮件、钉钉、打电话

　　2）主要的解决方案是尝试重启、然后查看日志定位问题

十、Spark Core & SQL

10.1 Spark有几种部署方式？请分别简要论述

　　1）Local本地模式，用于测试

　　2）Standalone模式，Spark自带的一个调度系统

　　3）Yarn模式，Spark客户端直接连接Yarn，不需要额外构建Spark集群；有yarn-client和yarn-cluster两个模式，主要区别在于Driver程序的运行节点

　　4）Mesos模式，国内用的少

10.2 Spark任务使用什么进行提交？Java界面还是脚本？

　　shell脚本

10.3 Spark提交作业参数

　　1）executor-cores，每个executor使用的内核数，默认为1，官方建议2-5个，我们企业是4个

　　2）num-executors，启动executors的数量，默认为2

　　3）executor-memory，executor内存大小，默认1G

　　4）driver-cores， driver使用内核数，默认为1

　　5）driver-memory，driver内存大小，默认512M

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

10.4 简述Spark的架构与作业提交流程（画图讲解，注明各个部分的作用）

10.4.1 SparkOnYarnClient提交流程

　　1）客户端提交应用给ResourceManager进行处理，并且在Driver端启动一个DriverEndpoint

　　2）ResourceManager收到请求后选择一个NodeManager来启动AppMaster

　　3）ApppMaster启动后向ResourceManager请求注册自己，表示自己已经启动，并向ResourceManager申请资源运行Task任务

　　4）ResourceManager获取请求后给AppMaster分配资源，AppMaster获取资源后开始选择一个NodeManager来启动Container容器

　　5）容器启动后有YarnCoarseGrained ExecutorBackend跟Driver端的DriverEndpoint进行反向注册

　　6）DriverEndpoint收到注册请求后返回注册完成响应给YarnCoarseGrained ExecutorBackend

　　7）YarnCoarseGrained ExecutorBackend开始创建执行器，创建Executor

　　8）Executor创建成功后YarnCoarseGrained ExecutorBackend发送通知给DriverEndpoint，说：Executor已经创建成功

　　9）DriverEndpoint随后给YarnCoarseGrained ExecutorBackend发送offset，执行Task任务

10.4.2 SparkOnYarnCluster提交流程

　　1）客户端提交应用给ResourceManager进行处理

　　2）ResourceManager收到请求后选择一个NodeManager来启动AppMaster，并且在AppMaster中启动Driver

　　3）ApppMaster启动后向ResourceManager请求注册自己，表示自己已经启动，并向ResourceManager申请资源运行Task任务

　　4）ResourceManager获取请求后给AppMaster分配资源，AppMaster获取资源后开始选择一个NodeManager来启动Container容器

　　5）容器启动后有YarnCoarseGrained ExecutorBackend跟Driver中的DriverEndpoint进行反向注册

　　6）DriverEndpoint收到注册请求后返回注册完成响应给YarnCoarseGrained ExecutorBackend

　　7）YarnCoarseGrained ExecutorBackend开始创建执行器，创建Executor

　　8）Executor创建成功后YarnCoarseGrained ExecutorBackend发送通知给DriverEndpoint，说：Executor已经创建成功

　　9）DriverEndpoint随后给YarnCoarseGrained ExecutorBackend发送offset，执行Task任务

10.5 如何理解Spark中血统的概念（RDD）？

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

10.6 简述Spark的宽窄依赖，以及Spark如何划分Stage，每个Stage又根据什么决定Task个数

　　宽窄依赖可以用来解决数据容错时的高效性以及参与任务的划分，不同的是窄依赖没有shuffle操作，宽依赖有shuffle操作；Spark根据RDD之间的依赖关系将不同的Job划分为不同的Stage，一个宽依赖划分一个Stage；每个Stage根据分区数划分Task，Stage就是一个TaskSet集合

10.7 请举例Spark的Transformation算子和Action算子，并简述功能

10.7.1 Transformation算子

　　1）map()：输入一个rdd，输出一个新rdd

　　2）mapPartitions()：以分区为单位进行map()操作

　　3）flatMap()：扁平化操作

　　4）groupBy()：分组

　　5）filter()：过滤

　　6）sample()：采样

　　7）coalesce()：加减分区；当进行缩减分区时，第二个参数为true时，会进行shuffle操作

　　8）repartition()：重分区

　　9）sortBy()：排序

　　10）intersection()：取交集

　　11）union()：取并集

　　12）subtract()：取差集

　　13）zip()：制作拉链表

　　14）partitionBy()：按照key重分区

　　15）reduceByKey()：按照key聚合value，溢写磁盘时进行了类似于MR的Combiner预聚合，效率比groupBy + map高

　　16）gourpByKey()：按照key分组

　　17）sortByKey()：按照key排序

　　18）mapValues()：只对V进行操作

　　19）join()

　　20）aggregateByKey()：按照key处理分区内核分区间的逻辑

　　21）foldByKey()：分区内和分区间逻辑相同的aggregateByKey()

　　22）combineByKey()：转换结构后分区内和分区间的操作

10.7.2 Action算子

　　1）reduce()：聚合

　　2）collect()：以数据形式返回数据集，所有数据会被拉取到Driver端，尽量不用

　　3）first()：返回rdd中的第一个元素

　　4）count()：返回rdd中元素的个数

　　5）take()：返回由rdd前n个元素组成的数组

　　6）takeOrdered()：返回rdd排序后前n个元素组成的数组

　　7）countByKey()：按key统计key的个数

　　8）foreach()：遍历rdd中的每一个元素

　　9）saveAsTextFile(path)保存成Text文件

　　10）saveAsSequenceFile(path) 保存成Sequencefile文件（只有kv类型RDD有该操作，单值的没有）

　　11）saveAsObjectFile(path) 序列化成对象保存到文件

10.8 请举例会引起Shuffle过程的Spark算子，并简述功能

　　1）groupBy()：分组

　　2）coalesce()：加减分区；当进行缩减分区时，第二个参数为true时，会进行shuffle操作

　　3）repartition()：重分区

　　4）subtract()：取差集

　　5）gourpByKey()：按照key分组

　　6）reduceByKey()：按照key聚合value，溢写磁盘时进行了类似于MR的Combiner预聚合，效率比groupBy + map高

　　7）aggregateByKey()：按照key处理分区内核分区间的逻辑

　　8）foldByKey()：分区内和分区间逻辑相同的aggregateByKey()

　　9）combineByKey()：转换结构后分区内和分区间的操作

　　10）sortByKey()：按照key排序

10.9 简述Spark的两个核心Shuffle的工作流程，包括未优化的和优化的HashShuffle，普通的SortShuffle和Bypass的SortShuffle

 10.9.1 未经优化的HashShuffle

　　不同Task中Key相同的数据不会在Executor中提前进行聚合，而是每个Task各自输出自己的数据

 10.9.2 优化后的HashShuffle

　　同一个Executor中的不同Task产生的数据可以根据相同的Key提前进行聚合，这样在shuffle过程中就能大大减少网络IO的压力

 10.9.3 普通的SortShuffle

 10.9.1 bypass中的SortShuffle

　　当 shuffle read task 的 数 量 小 于 等 于 spark.shuffle.sort。bypassMergeThreshold 参数的值时（默认为 200），就会启用 bypass 机制

10.10 Spark常用算子ReduceByKey与GroupByKey的区别，哪一种更具优势？

　　reduceByKey：按照key进行聚合，在shuffle之前有combine（预聚合）操作，但是需要注意是否会影响业务逻辑，性能比groupByKey高；

　　groupByKey：按照key进行分组，直接进行shuffle；

10.11 Repartition和Coalesce的关系与区别？

　　1）关系：两者都是用来改变rdd的分区数的，repartition底层调用的就是coalesce

　　2）区别：repartition一定发生shuffle，coalesce根据传入的参数判断是否会发生shuffle；一般情况下增大rdd分区数使用repartition，减少rdd分区数使用coalesce；当然coalesce也能增大分区数；

10.12 分别简述Spark中的缓存机制与Checkpoint机制，并指出两者的区别与联系

　　1）联系：都是做rdd持久化的

　　2）缓存机制：不会截断血缘关系，计算过程中优先使用数据缓存

　　3）checkpoint：磁盘，截断血缘关系，在ck之前必须没有任何任务提交才会生效，ck过程会额外提交一次任务

10.13 简述Spark中共享变量的基本原理与用途

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

　　共享变量出现的原因：通常在向Spark传递函数时，比如使用map()函数或者filter()传条件时，可以使用驱动器中定义的变量，但集群中运行的每个任务都会得到这些变量的一份新的副本，更新这些副本的值也不会影响驱动器中的对应变量。

　　Spark的两个共享变量，累加器和广播变量，分别为结果聚合与广播这两种常见的通信模式突破了这一限制。

10.14 当Spark涉及到数据库的操作时，如何减少Spark运行中的数据库连接数？

　　使用foreachPartition代替foreach，在foreachPartition内获取数据库的连接

10.15 如何使用Spark实现TopN的获取（描述思路或伪代码）？

10.15.1 方法一

　　1）按Key对数据进行聚合，groupByKey

　　2）将value转为数组，利用scala的sortBy或者sortWith进行排序，但是可能数据量比较大，会产生OOM

10.15.2 方法二

　　1）取出所有Key

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

10.15.3 方法三

　　1）自定义分区器，按照key进行分区，使不同的key进入不同的分区

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

10.16 调优之前与调优之后的性能对比

　　这里举个例子。比如我们有几百个文件，会有几百个map出现，读取之后进行join操作，会非常的慢。这个时候我们使用coalesce减少分区，比如240个map，我们合成60个map，也就是窄依赖。这样再shuffle，过程产生的文件数会大大减少。提高join的时间性能。

10.17 简述SparkSQL中RDD、DataFrame、DataSet三者之间的区别与联系？

10.18 append和overwrite的区别

　　append在原有分区上进行追加，overwrite在原有分区上进行全量刷新

10.19 coalesce和repartition的区别

　　coalesce和repartition都用于改变分区，coalesce用于缩小分区且不会进行shuffle，repartition用于增大分区（提供并行度）会进行shuffle,在spark中减少文件个数会使用coalesce来减少分区来到这个目的。但是如果数据量过大，分区数过少会出现OOM所以coalesce缩小分区个数也需合理

10.20 cache缓存级别

　　DataFrame的cache默认采用 MEMORY_AND_DISK 这和RDD 的默认方式不一样RDD cache 默认采用MEMORY_ONLY

10.21 释放缓存和缓存

　　缓存：(1)dataFrame.cache  (2)sparkSession.catalog.cacheTable(“tableName”)

　　释放缓存：(1)dataFrame.unpersist  (2)sparkSession.catalog.uncacheTable(“tableName”)

10.22 Spark Shuffle默认并行度

　　spark.sql.shuffle.partitions 决定 默认并行度200

10.23 kryo序列化

　　kryo序列化比java序列化更快更紧凑，但spark默认的序列化是java序列化并不是spark序列化，因为spark并不支持所有序列化类型，而且每次使用都必须进行注册。注册只针对于RDD。在DataFrames和DataSet当中自动实现了kryo序列化

10.24 创建临时表和全局临时表

　　DataFrame.createTempView() 创建普通临时表

　　DataFrame.createGlobalTempView()、DataFrame.createOrReplaceTempView() 创建全局临时表

10.25 BroadCast Join 

　　原理：先将小表数据查询出来聚合到driver端，再广播到各个executor端，使表与表join时进行本地join，避免进行网络传输产生shuffle。

　　使用场景：大表join小表 只能广播小表

10.26 控制Spark Reduce缓存，调优shuffle

　　spark.reducer.maxSizeInFilght 此参数为reduce task能够拉取多少数据量的一个参数默认48MB，当集群资源足够时，增大此参数可减少reduce拉取数据量的次数，从而达到优化shuffle的效果，一般调大为96MB,资源够大可继续往上跳。

　　spark.shuffle.file.buffer 此参数为每个shuffle文件输出流的内存缓冲区大小，调大此参数可以减少在创建shuffle文件时进行磁盘搜索和系统调用的次数，默认参数为32k 一般调大为64k

10.27 注册UDF函数

　　SparkSession.udf.register 方法进行注册

10.28 SparkSQL中Join操作与Left Join的区别？

　　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 参与关联计算了

十一、Spark Streaming

11.1 Spark Streaming第一次运行不丢失数据？

　　kafka参数 auto.offset.reset 参数设置成earliest 从最初始偏移量开始消费数据

11.2 Spark Streaming精准一次消费

　　手动维护偏移量，处理完业务数据后，再进行提交偏移量操作，极端情况下，如在提交偏移量时断网或停电会造成spark程序第二次启动时重复消费问题，所以在涉及到金额或精确性非常高的场景会使用事物保证精准一次消费

11.3 Spark Streaming控制每秒消费数据的速度

　　通过spark.streaming.kafka.maxRatePerPartition参数来设置Spark Streaming从kafka分区每秒拉取的条数

11.4 Spark Streaming背压机制

　　把spark.streaming.backpressure.enabled 参数设置为ture,开启背压机制后Spark Streaming会根据延迟动态去kafka消费数据,上限由spark.streaming.kafka.maxRatePerPartition参数控制，所以两个参数一般会一起使用

11.5 Spark Streaming 一个stage耗时

　　Spark Streaming stage耗时由最慢的task决定,所以数据倾斜时某个task运行慢会导致整个Spark Streaming都运行非常慢

11.6 Spark Streaming 优雅关闭

　　把spark.streaming.stopGracefullyOnShutdown参数设置成ture,Spark会在JVM关闭时正常关闭StreamingContext,而不是立马关闭

　　Kill 命令：yarn application -kill 后面跟 applicationid

11.7 Spark Streaming 默认分区个数

　　Spark Streaming默认分区个数与所对接的kafka topic分区个数一致，Spark Streaming里一般不会使用repartition算子增大分区，因为repartition会进行shuffle增加耗时

11.8 SparkStreaming有哪几种方式消费Kafka中的数据，它们之间的区别是什么？

11.8.1 基于Receiver的方式

　　这种方式使用Receiver来获取数据。Receiver是使用Kafka的高层次Consumer API来实现的。receiver从Kafka中获取的数据都是存储在Spark Executor的内存中的（如果突然数据暴增，大量batch堆积，很容易出现内存溢出的问题），然后Spark Streaming启动的job会去处理那些数据。 然而，在默认的配置下，这种方式可能会因为底层的失败而丢失数据。如果要启用高可靠机制，让数据零丢失，就必须启用Spark Streaming的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接收到的Kafka数据写入分布式文件系统（比如HDFS）上的预写日志中。所以，即使底层节点出现了失败，也可以使用预写日志中的数据进行恢复

11.8.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的副本进行恢复。
一次且仅一次的事务机制

11.8.3 对比

　　基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是却无法保证数据被处理一次且仅一次，可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。
　　基于direct的方式，使用kafka的简单api，Spark Streaming自己就负责追踪消费的offset，并保存在checkpoint中。Spark自己一定是同步的，因此可以保证数据是消费一次且仅消费一次。
　　在实际生产环境中大都用Direct方式

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

　　窗口函数就是在原来定义的SparkStreaming计算批次大小的基础上再次进行封装，每次计算多个批次的数据，同时还需要传递一个滑动步长的参数，用来设置当次计算任务完成之后下一次从什么地方开始计算。

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