[大数据]大数据岗位必备知识点总结

大数据岗位必备知识点总结

HDFS的读写流程？

• 写入数据

  • 1）客户端向namenode请求上传文件，namenode检查目标文件是否已存在，父目录是否存在。
  • 2）namenode返回是否可以上传。
  • 3）客户端请求第一个 block上传到哪几个datanode服务器上。
  • 4）namenode返回3个datanode节点，分别为dn1、dn2、dn3。
  • 5）客户端请求dn1上传数据，dn1收到请求会继续调用dn2，然后dn2调用dn3，将这个通信管道建立完成。
  • 6）dn1、dn2、dn3逐级应答客户端
  • 7）客户端开始往dn1上传第一个block（先从磁盘读取数据放到一个本地内存缓存），以packet为单位，dn1
  • 收到一个packet就会传给dn2，dn2传给dn3；dn1每传一个packet会放入一个应答队列等待应答
  • 8）当一个block传输完成之后，客户端再次请求namenode上传第二个block的服务器。（重复执行3-7步）

• 读取数据

  • 1）客户端向namenode请求下载文件，namenode通过查询元数据，找到文件块所在的datanode地址。
  • 2）挑选一台datanode（就近原则，然后随机）服务器，请求读取数据。
  • 3）datanode开始传输数据给客户端（从磁盘里面读取数据放入流，以packet为单位来做校验）。
  • 4）客户端以packet为单位接收，先在本地缓存，然后写入目标文件

Hive的架构

1.用户接口：Client

CLI(command-line interface)、JDBC/ODBC(jdbc访问hive)、WEBUI（浏览器访问hive）

2：元数据:Metastore

元数据包括：表名、表所属的数据库（默认是default）、表的拥有者、列/分区字段、表的类型（是否是外部表）、表的数据所在目录等；默认存储在自带的derby数据库中，推荐使用MySQL存储Metastore。

3:Hadoop

使用HDFS进行存储，使用MapReduce进行计算。

4:驱动器Driver

（1）解析器（SQL Parser）：将SQL字符串转换成抽象语法树AST，这一步一般都用第三方工具库完成，比如antlr；对AST进行语法分析，比如表是否存在、字段是否存在、SQL语义是否有误。

（2）编译器（Physical Plan）：将AST编译生成逻辑执行计划。

（3）优化器（Query Optimizer）：对逻辑执行计划进行优化。

（4）执行器（Execution）：把逻辑执行计划转换成可以运行的物理计划。对于Hive来说，就是MR/Spark。

Hive的使用和优化

1)MapJoin

如果不指定MapJoin或者不符合MapJoin的条件，那么Hive解析器会将Join操作转换成Common Join，即：在Reduce阶段完成join。容易发生数据倾斜。可以用MapJoin把小表全部加载到内存在map端进行join，避免reducer处理。

2）行列过滤

列处理：在SELECT中，只拿需要的列，如果有，尽量使用分区过滤，少用SELECT *。

行处理：在分区剪裁中，当使用外关联时，如果将副表的过滤条件写在Where后面，那么就会先全表关联，之后再过滤。

3）列式存储

4）采用分区技术

5）合理设置Map数

mapred.min.split.size: 指的是数据的最小分割单元大小；min的默认值是1B

mapred.max.split.size: 指的是数据的最大分割单元大小；max的默认值是256MB

通过调整max可以起到调整map数的作用，减小max可以增加map数，增大max可以减少map数。

需要提醒的是，直接调整mapred.map.tasks这个参数是没有效果的。

https://www.cnblogs.com/swordfall/p/11037539.html

6）合理设置Reduce数

Reduce个数并不是越多越好

（1）过多的启动和初始化Reduce也会消耗时间和资源；

（2）另外，有多少个Reduce，就会有多少个输出文件，如果生成了很多个小文件，那么如果这些小文件作为下一个任务的输入，则也会出现小文件过多的问题；

在设置Reduce个数的时候也需要考虑这两个原则：处理大数据量利用合适的Reduce数；使单个Reduce任务处理数据量大小要合适；

7）小文件如何产生的？

（1）动态分区插入数据，产生大量的小文件，从而导致map数量剧增；

（2）reduce数量越多，小文件也越多（reduce的个数和输出文件是对应的）；

（3）数据源本身就包含大量的小文件。

8）小文件解决方案

• （1）在Map执行前合并小文件，减少Map数：CombineHiveInputFormat具有对小文件进行合并的功能（系统默认的格式）。HiveInputFormat没有对小文件合并功能。
• (2）merge
• // 输出合并小文件
• SET hive.merge.mapfiles = true; – 默认true，在map-only任务结束时合并小文件
• SET hive.merge.mapredfiles = true; – 默认false，在map-reduce任务结束时合并小文件
• SET hive.merge.size.per.task = 268435456; – 默认256M
• SET hive.merge.smallfiles.avgsize = 16777216; – 当输出文件的平均大小小于16m该值时，启动一个独立的map-reduce任务进行文件merge
• （3）开启JVM重用
• set mapreduce.job.jvm.numtasks=10

9）开启map端combiner（不影响最终业务逻辑）

set hive.map.aggr=true；

10）压缩（选择快的）

• 设置map端输出、中间结果压缩。（不完全是解决数据倾斜的问题，但是减少了IO读写和网络传输，能提高很多效率）
• set hive.exec.compress.intermediate=true --启用中间数据压缩
• set mapreduce.map.output.compress=true --启用最终数据压缩
• set mapreduce.map.outout.compress.codec=…; --设置压缩方式

11）采用tez引擎或者spark引擎

Hive的常用函数和窗口函数?

• UDF-一进一出 upper

• UDAF 多进一出count

• UDTF 一进多出explode

• UDF（user-defined function）作用于单个数据行，产生一个数据行作为输出。（数学函数，字符串函数）

• UDAF（用户定义聚集函数 User- Defined Aggregation Funcation）：接收多个输入数据行，并产生一个输出数据行。（count，max）

• UDTF（表格生成函数 User-Defined Table Functions）：接收一行输入，输出（explode）

• 常用函数？

• 系统函数

• 1）date_add、date_sub函数（加减日期）

• 2）next_day函数（周指标相关）

• 3）date_format函数（根据格式整理日期）

• 4）last_day函数（求当月最后一天日期）

• 5）collect_set函数

• 6）get_json_object解析json函数

• 7）NVL（表达式1，表达式2）

• 如果表达式1为空值，NVL返回值为表达式2的值，否则返回表达式1的值。

• 窗口函数？

• 1）Rank

• （1）RANK() 排序相同时会重复，总数不会变

• （2）DENSE_RANK() 排序相同时会重复，总数会减少

• （3）ROW_NUMBER() 会根据顺序计算

• （ 2） OVER()：指定分析函数工作的数据窗口大小，这个数据窗口大小可能会随着行的变而变化

• （1）CURRENT ROW：当前行current

• （2）n PRECEDING：往前n行数据 percending

• （3） n FOLLOWING：往后n行数据following

• （4）UNBOUNDED(unbounded)：起点，UNBOUNDED PRECEDING 表示从前面的起点， UNBOUNDED FOLLOWING表示到后面的终点

• （5） LAG(col,n)：往前第n行数据

• （6）LEAD(col,n)：往后第n行数据

• （7） NTILE(n)：把有序分区中的行分发到指定数据的组中，各个组有编号，编号从1开始，对于每一行，NTILE返回此行所属的组的编号。注意：n必须为int类型。

Spark 任务提交流程？

• 1.yarn 集群，NM向RM 报送心跳，RM等待应用向自己提交申请。
• 2.在一台Linux上运行spark-submit 命令 向yarn申请资源，
• 3.客户端向RM 申请启动Appclication Master
• .4. 在不忙的NM 的一个Container 资源容器中启动APPMaster
• 5.AM 向RM 申请资源用于Executor，然后返回可用的Executor，AM连接NM,在Container中启动Executor，按照申请，启动Executor,然后反向注册给Driver.
• 6 Driver接收到所有的executor 都启动完毕
• 7.用户创建sparkcontext 自动创建DAGScheduler 和TaskScheduler
• 8 SparkContext 识别到saveASTextFile 是一个action算子，这是一个算子，回溯前面的RDD的依赖关系构建成DAG图
• 9.DAGScheduler 将DAG图依据宽依赖shuffle 算子，DAG 分解多个Stage,RDD之间是窄依赖的就合在一个stage内，一个Stage就是一个TaskSet,一个分区对应一个Task,TaskSet包含一组Task.
• 10.DAGScheduler 将包含的一组task的stage 提交给TaskScheduler。TaskScheduler 将Task发送 给不同的Executor
• 11.Executor 将Task放到线程池中执行。执行完成后把调度信息反馈给任务调用器
• 12 Stage1执行完成后 ，继续反馈给DAG调度器，然后DAG继续把stage2的Taskset给DAGScheduler 处理，然后重复以上步骤
• 13。等main方法所有任务执行完成后，会销毁所有的Driver和Executor 然后RM继续等待别的应用向他申请资源。

Spark任务划分和任务调度策略？

• 任务划分
• RDD任务切分中间分为：Application、Job、Stage和Task
• 1）Application：初始化一个SparkContext即生成一个Application；
• 2）Job：一个Action算子就会生成一个Job；
• 3）Stage：根据RDD之间的依赖关系的不同将Job划分成不同的Stage，遇到一个宽依赖则划分一个Stage；
• 对于宽依赖，由于有Shuffle的存在，只能在parent RDD处理完成后，才能开始接下来的计算，因此宽依赖是划分Stage的依据。
• 4）Task：Stage是一个TaskSet，将Stage划分的结果发送到不同的Executor执行即为一个Task。
• 任务调度
• Spark中的调度模式主要有两种：FIFO和FAIR。 默认情况下Spark的调度模式是FIFO（先进先出），谁先提交谁先执行，后面的任务需要等待前面的任务执行。 而FAIR（公平调度）模式支持在调度池中为任务进行分组，不同的调度池权重不同，任务可以按照权重来决定执行顺序。

Spark 的Shuffle机制？

在Spark中，Shuffle分为map阶段和reduce阶段，也可称之为shuffle write和shuffle read阶段，Spark在1.1以前的版本一直是采用Hash Shuffle的实现的方式，到1.1版本时参考Hadoop MapReduce的实现开始引入Sort Shuffle，在1.5版本时开始Tungsten钨丝计划，引入UnSafe Shuffle优化内存及CPU的使用，在1.6中将Tungsten统一到Sort Shuffle中，实现自我感知选择最佳Shuffle方式，到最近的2.0版本， Hash Shuffle已被删除，所有Shuffle方式全部统一到Sort Shuffle一个实现中。这里我们只说明Sort Shuffle

Write 阶段会把 Mapper 中每个 MapTask 所有的输出数据排好序，然后写到一个 Data 文件中， 同时还会写一份 index 文件用于记录 Data 文件内部分区数据的元数据（即记录哪一段数据是输出给哪个 ReduceTask 的），所以 Mapper 中的每一个 MapTask 会产生两个文件 。

Read 阶段首先 Reducer 会找 Driver 去获取父 Stage 中每个 MapTask 输出的位置信息，跟据位置信息获取并解析 Index 文件，会根据 Index 文件记录的信息来获取所需要读取的 Data 文件中属于自己那一部分的数据。

*Mapper 端的排序分为两个部分：内部排序和分区排序，每个分区内部的数据是有序的，分区之间也是有序的，

如图Data 文件中有三个分区，分别着对应下游的 ReduceTask，分区排序的目的是让 ReduceTask

获取数据更加高效

*Sorted-Based Shuffle 过程 Mapper 端生成的文件个数与什么有关？ 一个 core 即一个 Task

即一个分区即一个并发，一个 Task 会生成两个文件，Data 文件和 Index 文件。与 Reduce 端无关，

Reduce 只负责读属于自己的那一部分数据。

*Sorted-Based Shuffle解决了Hash Shuffle在Mapper端做ShuffleWriter时所产生的大量文件问题，

降低了资源消耗

*当前的Shuffle Write有三种 BypassMergeSortShuffleWriter， SortShuffleWriter 和

UnsafeShuffleWriter.

触发Shuffle的一些操作

Spark性能调优经验。

• 原则一：避免创建重复的RDD

• 通常来说，我们在开发一个Spark作业时，首先是基于某个数据源（比如Hive表或HDFS文件）创建一个初始的RDD；接着对这个RDD执行某个算子操作，然后得到下一个RDD；以此类推，循环往复，直到计算出最终我们需要的结果。在这个过程中，多个RDD会通过不同的算子操作（比如map、reduce等）串起来，这个“RDD串”，就是RDD lineage，也就是“RDD的血缘关系链”。

• 我们在开发过程中要注意：对于同一份数据，只应该创建一个RDD，不能创建多个RDD来代表同一份数据。

• 一些Spark初学者在刚开始开发Spark作业时，或者是有经验的工程师在开发RDD lineage极其冗长的Spark作业时，可能会忘了自己之前对于某一份数据已经创建过一个RDD了，从而导致对于同一份数据，创建了多个RDD。这就意味着，我们的Spark作业会进行多次重复计算来创建多个代表相同数据的RDD，进而增加了作业的性能开销。

• 一个简单的例子

• // 需要对名为“hello.txt”的HDFS文件进行一次map操作，再进行一次reduce操作。也就是说，需要对一份数据执行两次算子操作。// 错误的做法：对于同一份数据执行多次算子操作时，创建多个RDD。// 这里执行了两次textFile方法，针对同一个HDFS文件，创建了两个RDD出来，然后分别对每个RDD都执行了一个算子操作。// 这种情况下，Spark需要从HDFS上两次加载hello.txt文件的内容，并创建两个单独的RDD；第二次加载HDFS文件以及创建RDD的性能开销，很明显是白白浪费掉的。val rdd1 = sc.textFile(“hdfs://192.168.0.1:9000/hello.txt”)rdd1.map(…)val rdd2 = sc.textFile(“hdfs://192.168.0.1:9000/hello.txt”)rdd2.reduce(…)// 正确的用法：对于一份数据执行多次算子操作时，只使用一个RDD。// 这种写法很明显比上一种写法要好多了，因为我们对于同一份数据只创建了一个RDD，然后对这一个RDD执行了多次算子操作。// 但是要注意到这里为止优化还没有结束，由于rdd1被执行了两次算子操作，第二次执行reduce操作的时候，还会再次从源头处重新计算一次rdd1的数据，因此还是会有重复计算的性能开销。// 要彻底解决这个问题，必须结合“原则三：对多次使用的RDD进行持久化”，才能保证一个RDD被多次使用时只被计算一次。val rdd1 = sc.textFile(“hdfs://192.168.0.1:9000/hello.txt”)rdd1.map(…)rdd1.reduce(…)

• 原则二：尽可能复用同一个RDD

• 除了要避免在开发过程中对一份完全相同的数据创建多个RDD之外，在对不同的数据执行算子操作时还要尽可能地复用一个RDD。比如说，有一个RDD的数据格式是key-value类型的，另一个是单value类型的，这两个RDD的value数据是完全一样的。那么此时我们可以只使用key-value类型的那个RDD，因为其中已经包含了另一个的数据。对于类似这种多个RDD的数据有重叠或者包含的情况，我们应该尽量复用一个RDD，这样可以尽可能地减少RDD的数量，从而尽可能减少算子执行的次数。

• 一个简单的例子

• // 错误的做法。// 有一个<Long, String>格式的RDD，即rdd1。// 接着由于业务需要，对rdd1执行了一个map操作，创建了一个rdd2，而rdd2中的数据仅仅是rdd1中的value值而已，也就是说，rdd2是rdd1的子集。JavaPairRDD<Long, String> rdd1 = …JavaRDD rdd2 = rdd1.map(…)// 分别对rdd1和rdd2执行了不同的算子操作。rdd1.reduceByKey(…)rdd2.map(…)// 正确的做法。// 上面这个case中，其实rdd1和rdd2的区别无非就是数据格式不同而已，rdd2的数据完全就是rdd1的子集而已，却创建了两个rdd，并对两个rdd都执行了一次算子操作。// 此时会因为对rdd1执行map算子来创建rdd2，而多执行一次算子操作，进而增加性能开销。// 其实在这种情况下完全可以复用同一个RDD。// 我们可以使用rdd1，既做reduceByKey操作，也做map操作。// 在进行第二个map操作时，只使用每个数据的tuple._2，也就是rdd1中的value值，即可。JavaPairRDD<Long, String> rdd1 = …rdd1.reduceByKey(…)rdd1.map(tuple._2…)// 第二种方式相较于第一种方式而言，很明显减少了一次rdd2的计算开销。// 但是到这里为止，优化还没有结束，对rdd1我们还是执行了两次算子操作，rdd1实际上还是会被计算两次。// 因此还需要配合“原则三：对多次使用的RDD进行持久化”进行使用，才能保证一个RDD被多次使用时只被计算一次。

• 原则三：对多次使用的RDD进行持久化

• 当你在Spark代码中多次对一个RDD做了算子操作后，恭喜，你已经实现Spark作业第一步的优化了，也就是尽可能复用RDD。此时就该在这个基础之上，进行第二步优化了，也就是要保证对一个RDD执行多次算子操作时，这个RDD本身仅仅被计算一次。

• Spark中对于一个RDD执行多次算子的默认原理是这样的：每次你对一个RDD执行一个算子操作时，都会重新从源头处计算一遍，计算出那个RDD来，然后再对这个RDD执行你的算子操作。这种方式的性能是很差的。

• 因此对于这种情况，我们的建议是：对多次使用的RDD进行持久化。此时Spark就会根据你的持久化策略，将RDD中的数据保存到内存或者磁盘中。以后每次对这个RDD进行算子操作时，都会直接从内存或磁盘中提取持久化的RDD数据，然后执行算子，而不会从源头处重新计算一遍这个RDD，再执行算子操作。

• 对多次使用的RDD进行持久化的代码示例

• // 如果要对一个RDD进行持久化，只要对这个RDD调用cache()和persist()即可。// 正确的做法。// cache()方法表示：使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。// 此时再对rdd1执行两次算子操作时，只有在第一次执行map算子时，才会将这个rdd1从源头处计算一次。// 第二次执行reduce算子时，就会直接从内存中提取数据进行计算，不会重复计算一个rdd。val rdd1 = sc.textFile(“hdfs://192.168.0.1:9000/hello.txt”).cache()rdd1.map(…)rdd1.reduce(…)// persist()方法表示：手动选择持久化级别，并使用指定的方式进行持久化。// 比如说，StorageLevel.MEMORY_AND_DISK_SER表示，内存充足时优先持久化到内存中，内存不充足时持久化到磁盘文件中。// 而且其中的_SER后缀表示，使用序列化的方式来保存RDD数据，此时RDD中的每个partition都会序列化成一个大的字节数组，然后再持久化到内存或磁盘中。// 序列化的方式可以减少持久化的数据对内存/磁盘的占用量，进而避免内存被持久化数据占用过多，从而发生频繁GC。val rdd1 = sc.textFile(“hdfs://192.168.0.1:9000/hello.txt”).persist(StorageLevel.MEMORY_AND_DISK_SER)rdd1.map(…)rdd1.reduce(…)

• 对于persist()方法而言，我们可以根据不同的业务场景选择不同的持久化级别。

• Spark的持久化级别

• 持久化级别含义解释

• MEMORY_ONLY 使用未序列化的Java对象格式，将数据保存在内存中。如果内存不够存放所有的数据，则数据可能就不会进行持久化。那么下次对这个RDD执行算子操作时，那些没有被持久化的数据，需要从源头处重新计算一遍。这是默认的持久化策略，使用cache()方法时，实际就是使用的这种持久化策略。

• MEMORY_AND_DISK 使用未序列化的Java对象格式，优先尝试将数据保存在内存中。如果内存不够存放所有的数据，会将数据写入磁盘文件中，下次对这个RDD执行算子时，持久化在磁盘文件中的数据会被读取出来使用。

• MEMORY_ONLY_SER 基本含义同MEMORY_ONLY。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC。

• MEMORY_AND_DISK_SER 基本含义同MEMORY_AND_DISK。唯一的区别是，会将RDD中的数据进行序列化，RDD的每个partition会被序列化成一个字节数组。这种方式更加节省内存，从而可以避免持久化的数据占用过多内存导致频繁GC。

• DISK_ONLY 使用未序列化的Java对象格式，将数据全部写入磁盘文件中。

• MEMORY_ONLY_2, MEMORY_AND_DISK_2, 等等. 对于上述任意一种持久化策略，如果加上后缀_2，代表的是将每个持久化的数据，都复制一份副本，并将副本保存到其他节点上。这种基于副本的持久化机制主要用于进行容错。假如某个节点挂掉，节点的内存或磁盘中的持久化数据丢失了，那么后续对RDD计算时还可以使用该数据在其他节点上的副本。如果没有副本的话，就只能将这些数据从源头处重新计算一遍了。

• 如何选择一种最合适的持久化策略

• 默认情况下，性能最高的当然是MEMORY_ONLY，但前提是你的内存必须足够足够大，可以绰绰有余地存放下整个RDD的所有数据。因为不进行序列化与反序列化操作，就避免了这部分的性能开销；对这个RDD的后续算子操作，都是基于纯内存中的数据的操作，不需要从磁盘文件中读取数据，性能也很高；而且不需要复制一份数据副本，并远程传送到其他节点上。但是这里必须要注意的是，在实际的生产环境中，恐怕能够直接用这种策略的场景还是有限的，如果RDD中数据比较多时（比如几十亿），直接用这种持久化级别，会导致JVM的OOM内存溢出异常。

• 如果使用MEMORY_ONLY级别时发生了内存溢出，那么建议尝试使用MEMORY_ONLY_SER级别。该级别会将RDD数据序列化后再保存在内存中，此时每个partition仅仅是一个字节数组而已，大大减少了对象数量，并降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销，主要就是序列化与反序列化的开销。但是后续算子可以基于纯内存进行操作，因此性能总体还是比较高的。此外，可能发生的问题同上，如果RDD中的数据量过多的话，还是可能会导致OOM内存溢出的异常。

• 如果纯内存的级别都无法使用，那么建议使用MEMORY_AND_DISK_SER策略，而不是MEMORY_AND_DISK策略。因为既然到了这一步，就说明RDD的数据量很大，内存无法完全放下。序列化后的数据比较少，可以节省内存和磁盘的空间开销。同时该策略会优先尽量尝试将数据缓存在内存中，内存缓存不下才会写入磁盘。

• 通常不建议使用DISK_ONLY和后缀为_2的级别：因为完全基于磁盘文件进行数据的读写，会导致性能急剧降低，有时还不如重新计算一次所有RDD。后缀为_2的级别，必须将所有数据都复制一份副本，并发送到其他节点上，数据复制以及网络传输会导致较大的性能开销，除非是要求作业的高可用性，否则不建议使用。

• 原则四：尽量避免使用shuffle类算子

• 如果有可能的话，要尽量避免使用shuffle类算子。因为Spark作业运行过程中，最消耗性能的地方就是shuffle过程。shuffle过程，简单来说，就是将分布在集群中多个节点上的同一个key，拉取到同一个节点上，进行聚合或join等操作。比如reduceByKey、join等算子，都会触发shuffle操作。

• shuffle过程中，各个节点上的相同key都会先写入本地磁盘文件中，然后其他节点需要通过网络传输拉取各个节点上的磁盘文件中的相同key。而且相同key都拉取到同一个节点进行聚合操作时，还有可能会因为一个节点上处理的key过多，导致内存不够存放，进而溢写到磁盘文件中。因此在shuffle过程中，可能会发生大量的磁盘文件读写的IO操作，以及数据的网络传输操作。磁盘IO和网络数据传输也是shuffle性能较差的主要原因。

• 因此在我们的开发过程中，能避免则尽可能避免使用reduceByKey、join、distinct、repartition等会进行shuffle的算子，尽量使用map类的非shuffle算子。这样的话，没有shuffle操作或者仅有较少shuffle操作的Spark作业，可以大大减少性能开销。

• Broadcast与map进行join代码示例

• // 传统的join操作会导致shuffle操作。// 因为两个RDD中，相同的key都需要通过网络拉取到一个节点上，由一个task进行join操作。val rdd3 = rdd1.join(rdd2)// Broadcast+map的join操作，不会导致shuffle操作。// 使用Broadcast将一个数据量较小的RDD作为广播变量。val rdd2Data = rdd2.collect()val rdd2DataBroadcast = sc.broadcast(rdd2Data)// 在rdd1.map算子中，可以从rdd2DataBroadcast中，获取rdd2的所有数据。// 然后进行遍历，如果发现rdd2中某条数据的key与rdd1的当前数据的key是相同的，那么就判定可以进行join。// 此时就可以根据自己需要的方式，将rdd1当前数据与rdd2中可以连接的数据，拼接在一起（String或Tuple）。val rdd3 = rdd1.map(rdd2DataBroadcast…)// 注意，以上操作，建议仅仅在rdd2的数据量比较少（比如几百M，或者一两G）的情况下使用。// 因为每个Executor的内存中，都会驻留一份rdd2的全量数据。

• 原则五：使用map-side预聚合的shuffle操作

• 如果因为业务需要，一定要使用shuffle操作，无法用map类的算子来替代，那么尽量使用可以map-side预聚合的算子。

• 所谓的map-side预聚合，说的是在每个节点本地对相同的key进行一次聚合操作，类似于MapReduce中的本地combiner。map-side预聚合之后，每个节点本地就只会有一条相同的key，因为多条相同的key都被聚合起来了。其他节点在拉取所有节点上的相同key时，就会大大减少需要拉取的数据数量，从而也就减少了磁盘IO以及网络传输开销。通常来说，在可能的情况下，建议使用reduceByKey或者aggregateByKey算子来替代掉groupByKey算子。因为reduceByKey和aggregateByKey算子都会使用用户自定义的函数对每个节点本地的相同key进行预聚合。而groupByKey算子是不会进行预聚合的，全量的数据会在集群的各个节点之间分发和传输，性能相对来说比较差。

• 比如如下两幅图，就是典型的例子，分别基于reduceByKey和groupByKey进行单词计数。其中第一张图是groupByKey的原理图，可以看到，没有进行任何本地聚合时，所有数据都会在集群节点之间传输；第二张图是reduceByKey的原理图，可以看到，每个节点本地的相同key数据，都进行了预聚合，然后才传输到其他节点上进行全局聚合。

• 原则六：使用高性能的算子

• 除了shuffle相关的算子有优化原则之外，其他的算子也都有着相应的优化原则。

• 使用reduceByKey/aggregateByKey替代groupByKey

• 详情见“原则五：使用map-side预聚合的shuffle操作”。

• 使用mapPartitions替代普通map

• mapPartitions类的算子，一次函数调用会处理一个partition所有的数据，而不是一次函数调用处理一条，性能相对来说会高一些。但是有的时候，使用mapPartitions会出现OOM（内存溢出）的问题。因为单次函数调用就要处理掉一个partition所有的数据，如果内存不够，垃圾回收时是无法回收掉太多对象的，很可能出现OOM异常。所以使用这类操作时要慎重！

• 使用foreachPartitions替代foreach

• 原理类似于“使用mapPartitions替代map”，也是一次函数调用处理一个partition的所有数据，而不是一次函数调用处理一条数据。在实践中发现，foreachPartitions类的算子，对性能的提升还是很有帮助的。比如在foreach函数中，将RDD中所有数据写MySQL，那么如果是普通的foreach算子，就会一条数据一条数据地写，每次函数调用可能就会创建一个数据库连接，此时就势必会频繁地创建和销毁数据库连接，性能是非常低下；但是如果用foreachPartitions算子一次性处理一个partition的数据，那么对于每个partition，只要创建一个数据库连接即可，然后执行批量插入操作，此时性能是比较高的。实践中发现，对于1万条左右的数据量写MySQL，性能可以提升30%以上。

• 使用filter之后进行coalesce操作

• 通常对一个RDD执行filter算子过滤掉RDD中较多数据后（比如30%以上的数据），建议使用coalesce算子，手动减少RDD的partition数量，将RDD中的数据压缩到更少的partition中去。因为filter之后，RDD的每个partition中都会有很多数据被过滤掉，此时如果照常进行后续的计算，其实每个task处理的partition中的数据量并不是很多，有一点资源浪费，而且此时处理的task越多，可能速度反而越慢。因此用coalesce减少partition数量，将RDD中的数据压缩到更少的partition之后，只要使用更少的task即可处理完所有的partition。在某些场景下，对于性能的提升会有一定的帮助。

• 使用repartitionAndSortWithinPartitions替代repartition与sort类操作

• repartitionAndSortWithinPartitions是Spark官网推荐的一个算子，官方建议，如果需要在repartition重分区之后，还要进行排序，建议直接使用repartitionAndSortWithinPartitions算子。因为该算子可以一边进行重分区的shuffle操作，一边进行排序。shuffle与sort两个操作同时进行，比先shuffle再sort来说，性能可能是要高的。

• 原则七：广播大变量

• 有时在开发过程中，会遇到需要在算子函数中使用外部变量的场景（尤其是大变量，比如100M以上的大集合），那么此时就应该使用Spark的广播（Broadcast）功能来提升性能。

• 在算子函数中使用到外部变量时，默认情况下，Spark会将该变量复制多个副本，通过网络传输到task中，此时每个task都有一个变量副本。如果变量本身比较大的话（比如100M，甚至1G），那么大量的变量副本在网络中传输的性能开销，以及在各个节点的Executor中占用过多内存导致的频繁GC，都会极大地影响性能。

• 因此对于上述情况，如果使用的外部变量比较大，建议使用Spark的广播功能，对该变量进行广播。广播后的变量，会保证每个Executor的内存中，只驻留一份变量副本，而Executor中的task执行时共享该Executor中的那份变量副本。这样的话，可以大大减少变量副本的数量，从而减少网络传输的性能开销，并减少对Executor内存的占用开销，降低GC的频率。

• 广播大变量的代码示例

• // 以下代码在算子函数中，使用了外部的变量。// 此时没有做任何特殊操作，每个task都会有一份list1的副本。val list1 = …rdd1.map(list1…)// 以下代码将list1封装成了Broadcast类型的广播变量。// 在算子函数中，使用广播变量时，首先会判断当前task所在Executor内存中，是否有变量副本。// 如果有则直接使用；如果没有则从Driver或者其他Executor节点上远程拉取一份放到本地Executor内存中。// 每个Executor内存中，就只会驻留一份广播变量副本。val list1 = …val list1Broadcast = sc.broadcast(list1)rdd1.map(list1Broadcast…)

• 原则八：使用Kryo优化序列化性能

• 在Spark中，主要有三个地方涉及到了序列化： * 在算子函数中使用到外部变量时，该变量会被序列化后进行网络传输（见“原则七：广播大变量”中的讲解）。 * 将自定义的类型作为RDD的泛型类型时（比如JavaRDD，Student是自定义类型），所有自定义类型对象，都会进行序列化。因此这种情况下，也要求自定义的类必须实现Serializable接口。 * 使用可序列化的持久化策略时（比如MEMORY_ONLY_SER），Spark会将RDD中的每个partition都序列化成一个大的字节数组。

• 对于这三种出现序列化的地方，我们都可以通过使用Kryo序列化类库，来优化序列化和反序列化的性能。Spark默认使用的是Java的序列化机制，也就是ObjectOutputStream/ObjectInputStream API来进行序列化和反序列化。但是Spark同时支持使用Kryo序列化库，Kryo序列化类库的性能比Java序列化类库的性能要高很多。官方介绍，Kryo序列化机制比Java序列化机制，性能高10倍左右。Spark之所以默认没有使用Kryo作为序列化类库，是因为Kryo要求最好要注册所有需要进行序列化的自定义类型，因此对于开发者来说，这种方式比较麻烦。

• 以下是使用Kryo的代码示例，我们只要设置序列化类，再注册要序列化的自定义类型即可（比如算子函数中使用到的外部变量类型、作为RDD泛型类型的自定义类型等）：

• // 创建SparkConf对象。val conf = new SparkConf().setMaster(…).setAppName(…)// 设置序列化器为KryoSerializer。conf.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”)// 注册要序列化的自定义类型。conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

• 原则九：优化数据结构

• Java中，有三种类型比较耗费内存： * 对象，每个Java对象都有对象头、引用等额外的信息，因此比较占用内存空间。 * 字符串，每个字符串内部都有一个字符数组以及长度等额外信息。 * 集合类型，比如HashMap、LinkedList等，因为集合类型内部通常会使用一些内部类来封装集合元素，比如Map.Entry。

• 因此Spark官方建议，在Spark编码实现中，特别是对于算子函数中的代码，尽量不要使用上述三种数据结构，尽量使用字符串替代对象，使用原始类型（比如Int、Long）替代字符串，使用数组替代集合类型，这样尽可能地减少内存占用，从而降低GC频率，提升性能。

• 但是在笔者的编码实践中发现，要做到该原则其实并不容易。因为我们同时要考虑到代码的可维护性，如果一个代码中，完全没有任何对象抽象，全部是字符串拼接的方式，那么对于后续的代码维护和修改，无疑是一场巨大的灾难。同理，如果所有操作都基于数组实现，而不使用HashMap、LinkedList等集合类型，那么对于我们的编码难度以及代码可维护性，也是一个极大的挑战。因此笔者建议，在可能以及合适的情况下，使用占用内存较少的数据结构，但是前提是要保证代码的可维护性。

熟识Hbase的架构原理,能够整合Phoenix实现类Sql查询。

• Master

• HBase Master用于协调多个Region Server，侦测各个RegionServer之间的状态，并平衡RegionServer之间的负载。HBaseMaster还有一个职责就是负责分配Region给RegionServer。HBase允许多个Master节点共存，但是这需要Zookeeper的帮助。不过当多个Master节点共存时，只有一个Master是提供服务的，其他的Master节点处于待命的状态。当正在工作的Master节点宕机时，其他的Master则会接管HBase的集群。

• Region Server

• 对于一个RegionServer而言，其包括了多个Region。RegionServer的作用只是管理表格，以及实现读写操作。Client直接连接RegionServer，并通信获取HBase中的数据。对于Region而言，则是真实存放HBase数据的地方，也就说Region是HBase可用性和分布式的基本单位。如果当一个表格很大，并由多个CF组成时，那么表的数据将存放在多个Region之间，并且在每个Region中会关联多个存储的单元（Store）。

• Zookeeper

• 对于 HBase 而言，Zookeeper的作用是至关重要的。首先Zookeeper是作为HBase Master的HA解决方案。也就是说，是Zookeeper保证了至少有一个HBase Master 处于运行状态。并且Zookeeper负责Region和Region Server的注册。其实Zookeeper发展到目前为止，已经成为了分布式大数据框架中容错性的标准框架。不光是HBase，几乎所有的分布式大数据相关的开源框架，都依赖于Zookeeper实现HA。

Hbase的读写流程

• 1、HBase的读流程：

• （1）、HRegisonServer保存着meta表及数据表，首先client先访问zk，访问-ROOT-表，然后在zk上面获取.meta.表所在的位置信息，找到这个meta表在哪个HRegionServer上面保存着。

• （2）、接着client访问HRegionServer表从而读取.meta.进而获取.meta.表中存放的元数据。

• （3）、client通过.meta.中的元数据信息，访问对应的HRegionServer，然后扫描HRegionServer的Memstore和StoreFile来查询数据。

• （4）、最后把HRegionServer把数据反馈给client。

• 2、HBase的写流程：

• （1）、client访问zk中的-ROOT-表，然后后在访问.meta.表，并获取.meta.中的元数据。

• （2）、确定当前要写入的HRegion和HRegionServer。

• （3）、clinet向HRegionServer发出写相应的请求，HRegionServer收到请求并响应。

• （4）、client先将数据写入到HLog中，以防数据丢失。

• （5）、然后将数据写入到MemStore中。

• （6）、如果HLog和MemStore都写入成功了，那么表示这个条数据写入成功了。

• （7）、如果MemStore写入的数据达到了阈值，那么将会flush到StoreFile中。

• （8）、当StoreFile越来越多，会触发Compact合并操作，将过多的StoteFile合并成一个大的StoreFile。

• （9）、当StoreFile越来越多时，Region也会越来越大，当达到阈值时，会触发spilit操作，将这个Region一分为二。

熟识Flume基本组成，Source、Channel、Sink选型以及自定义拦截器的编写。

• 1）taildir source

（1）断点续传、多目录

（2）哪个Flume版本产生的？Apache1.7、CDH1.6

（3）没有断点续传功能时怎么做的？ 自定义

（4）taildir挂了怎么办？

不会丢数：断点续传

重复数据：

（5）怎么处理重复数据？

不处理：生产环境通常不处理，出现重复的概率比较低。处理会影响传输效率。

处理

? 自身：在taildirsource里面增加自定义事务，影响效率

? 找兄弟：下一级处理（hive dwd sparkstreaming flink布隆）、去重手段（groupby、开窗取窗口第一条、redis）

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

不支持。 自定义 递归遍历文件夹 + 读取文件

• 2）file channel /memory channel/kafka channel

（1）File Channel

? 数据存储于磁盘，优势：可靠性高；劣势：传输速度低

? 默认容量：100万event

注意：FileChannel可以通过配置dataDirs指向多个路径，每个路径对应不同的硬盘，增大Flume吞吐量。

（2）Memory Channel

? 数据存储于内存，优势：传输速度快；劣势：可靠性差

? 默认容量：100个event

（3）Kafka Channel

? 数据存储于Kafka，基于磁盘；

? 优势：可靠性高；

? 传输速度快 Kafka Channel 大于Memory Channel + Kafka Sink 原因省去了Sink阶段

（4）Kafka Channel哪个版本产生的？

Flume1.6 版本产生=》并没有火；因为有bug

? topic-start 数据内容

? topic-event 数据内容 ture 和false 很遗憾，都不起作用。

? 增加了额外清洗的工作量。

Flume1.7解决了这个问题，开始火了。

（5）生产环境如何选择

? 如果下一级是Kafka，优先选择Kafka Channel

? 如果是金融、对钱要求准确的公司，选择File Channel

? 如果就是普通的日志，通常可以选择Memory Channel

? 每天丢几百万数据 pb级 亿万富翁，掉1块钱会捡？

3）HDFS sink

（1）时间（1小时-2小时） or 大小128m、event个数（0禁止）

具体参数：hdfs.rollInterval=3600，hdfs.rollSize=134217728，hdfs.rollCount =0

4）事务

Source到Channel是Put事务

Channel到Sink是Take事务

Flume拦截器

1）拦截器注意事项

项目中自定义了：ETL拦截器。

采用两个拦截器的优缺点：优点，模块化开发和可移植性；缺点，性能会低一些

2）自定义拦截器步骤

（1）实现 Interceptor

（2）重写四个方法

? initialize 初始化

? public Event intercept(Event event) 处理单个Event

? public List intercept(List events) 处理多个Event，在这个方法中调用Event intercept(Event event)

? close方法

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

3）拦截器可以不用吗？

可以不用；需要在下一级hive的dwd层和sparksteaming里面处理

优势：只处理一次，轻度处理；劣势：影响性能，不适合做实时推荐这种对实时要求比较高的场景。

Flume Channel选择器

Replicating：默认选择器。功能：将数据发往下一级所有通道

Multiplexing：选择性发往指定通道。

Flume监控器

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

2）解决办法？

（1）自身：增加内存flume-env.sh 4-6g

-Xmx与-Xms最好设置一致，减少内存抖动带来的性能影响，如果设置不一致容易导致频繁fullgc。

（2）找朋友：增加服务器台数

搞活动 618 =》增加服务器=》用完在退出

日志服务器配置：8-16g内存、磁盘8T

Flume采集数据会丢失吗?（防止数据丢失的机制）

如果是FileChannel不会，Channel存储可以存储在File中，数据传输自身有事务。

如果是MemoryChannel有可能丢。

熟识Zookeeper选举机制，工作机制以及常用的操作命令。

半数机制：集群中半数以上机器存活，集群可用。所以****Zookeeper****适合安装奇数台服务器。

**** *Zookeeper*虽然在配置文件中并没有指定*Master*和*Slave*。但是，*Zookeeper*工作时，是有一个节点为*Leader*，其他则为****Follower*，*Leader****是通过内部的选举机制临时产生的。

新集群的选举方式？

1、服务器1*启动，此时只有它*一台服务器启动*了，它发出去的报文没有任何响应，所以它的*选举状态一直是LOOKING状态****。

2、服务器2*启动，它与*最开始启动的服务器1进行通信*，互相交换自己的选举结果，由于两者都没有历史数据，所以*id值较大的服务器2胜出*，但是由于没有达到超过*半数以上的服务器都同意*选举它(这个例子中的*半数以上是3*)，所以*服务器1、2还是继续保持LOOKING状态****。

3、服务器3*启动，根据前面的理论分析，服务器3成为服务器1、2、3中的老大，而与上面不同的是，*此时有三台服务器选举了它****，所以它成为了这次选举的Leader。

4、服务器4启动，根据前面的分析，理论上服务器4应该是服务器1、2、3、4中最大的，但是由于****前面已经有半数以上的服务器选举了服务器3****，所以它只能接收当小弟的命了。

5、服务器5启动，同4一样当小弟。

脑裂是怎产生的？

这就相当于原本一个集群，被分成了两个集群，出现了两个"大脑"，这就是所谓的"脑裂"现象。对于这种情况，其实也可以看出来，原本应该是统一的一个集群对外提供服务的，现在变成了两个集群同时对外提供服务，如果过了一会，断了的网络突然联通了，那么此时就会出现问题了，两个集群刚刚都对外提供服务了，数据该怎么合并，数据冲突怎么解决等等问题。刚刚在说明脑裂场景时有一个前提条件就是没有考虑过半机制，所以实际上Zookeeper集群中是不会轻易出现脑裂问题的，原因在于过半机制。

zookeeper的过半机制：在领导者选举的过程中，如果某台zkServer获得了超过半数的选票，则此zkServer就可以成为Leader了。举个简单的例子：如果现在集群中有5台zkServer，那么half=5/2=2，那么也就是说，领导者选举的过程中至少要有三台zkServer投了同一个zkServer，才会符合过半机制，才能选出来一个Leader。

Zookeeper 集群节点为什么要部署成奇数？

zookeeper容错指的是：当宕掉几个zookeeper节点服务器之后，剩下的个数必须大于宕掉的个数，也就是剩下的节点服务数必须大于n/2，这样zookeeper集群才可以继续使用，无论奇偶数都可以选举leader。例如5台zookeeper节点机器最多宕掉2台，还可以继续使用，因为剩下3台大于5/2。至于为什么最好为奇数个节点？这样是为了以最大容错服务器个数的条件下，能节省资源。比如，最大容错为2的情况下，对应的zookeeper服务数，奇数为5，而偶数为6，也就是6个zookeeper服务的情况下最多能宕掉2个服务，所以从节约资源的角度看，没必要部署6（偶数）个zookeeper服务节点。

zookeeper选举的过程中为什么一定要有一个过半机制验证？

因为这样不需要等待所有zkServer都投了同一个zkServer就可以选举出来一个Leader了，这样比较快，所以叫快速领导者选举算法。

zookeeper过半机制中为什么是大于，而不是大于等于？

这就是更脑裂问题有关系了，比如回到上文出现脑裂问题的场景 [如上图1]：当机房中间的网络断掉之后，机房1内的三台服务器会进行领导者选举，但是此时过半机制的条件是 “节点数 > 3”，也就是说至少要4台zkServer才能选出来一个Leader，所以对于机房1来说它不能选出一个Leader，同样机房2也不能选出一个Leader，这种情况下整个集群当机房间的网络断掉后，整个集群将没有Leader。而如果过半机制的条件是 “节点数 >= 3”，那么机房1和机房2都会选出一个Leader，这样就出现了脑裂。这就可以解释为什么过半机制中是大于而不是大于等于，目的就是为了防止脑裂。

熟识Kafka消息缓冲队列架构，分区分配策略 Kafka精准一次及优化。

• Kafka 如何保证消息的消费顺序？
• 在Kafka中Partition(分区)是真正保存消息的地方，发送的消息都存放在这里。Partition(分区)又存在于Topic（主题）中，并且一个Topic（主题）可以指定多个Partition(分区)。
• 在Kafka中，只保证Partition(分区)内有序，不保证Topic所有分区都是有序的。
• 所以 Kafka 要保证消息的消费顺序，可以有2种方法：
• 一、1个Topic（主题）只创建1个Partition(分区)，这样生产者的所有数据都发送到了一个Partition(分区)，保证了消息的消费顺序。
• 二、生产者在发送消息的时候指定要发送到哪个Partition(分区)。

熟识使用Sqoop将数据从Mysql到HDFS导入导出以及优化处理。

Sqoop导入导出Null存储一致性问题

Hive中的Null在底层是以“\N”来存储，而MySQL中的Null在底层就是Null，为了保证数据两端的一致性。在导出数据时采用–input-null-string和–input-null-non-string两个参数。导入数据时采用–null-string和–null-non-string。

熟识Java,Scala,Python语法，熟识面向对象和函数式编程思想，良好的代码编写规范。

于Topic（主题）中，并且一个Topic（主题）可以指定多个Partition(分区)。

• 在Kafka中，只保证Partition(分区)内有序，不保证Topic所有分区都是有序的。
• 所以 Kafka 要保证消息的消费顺序，可以有2种方法：
• 一、1个Topic（主题）只创建1个Partition(分区)，这样生产者的所有数据都发送到了一个Partition(分区)，保证了消息的消费顺序。
• 二、生产者在发送消息的时候指定要发送到哪个Partition(分区)。

熟识使用Sqoop将数据从Mysql到HDFS导入导出以及优化处理。

Sqoop导入导出Null存储一致性问题

Hive中的Null在底层是以“\N”来存储，而MySQL中的Null在底层就是Null，为了保证数据两端的一致性。在导出数据时采用–input-null-string和–input-null-non-string两个参数。导入数据时采用–null-string和–null-non-string。

熟识Java,Scala,Python语法，熟识面向对象和函数式编程思想，良好的代码编写规范。

熟识Linux开发环境，熟识常用Linux操作命令，简单的Shell脚本编写。