1. mapreduce

（1） mapreduce 流程

• 读取 HDFS 文件，每行解析成一个 <K,V> 对。
• 利用自定义的 map 函数，对 <K,V> 进行处理，转换为新 <K,V> 对输出。
• 对输出按任务定义的 Partitioner 和 Reducer 数进行分区。
• 对输出进行排序、分组。
• 对输出进行 combine（可选）。
• 将多个 map 任务的输出，按照不同的分区，通过网络 copy 到不同的 reduce 节点上。（shuffle）
• 对多个 map 的输出进行合并、排序。reduce 函数接收的是分组后的数据。
• 处理后，产生新的 <K,V> 输出，写入 HDFS 中。

(2) 如何实现

• topK
• 全排序
• 二次排序
• 大小表连接
• 大大表连接

（3） mapreduce shuffle

MapReduce 确保每个 reducer 的输入都按键排序。系统执行排序的过程——将 map 输出作为输入传给 reducer——称为 shuffle。

map 端

mapper 一般执行输入格式的解析、投影和过滤。map -> partition -> sort->spill（combine）->merge

缓存

map 产生输出时，利用缓存的方式写到内存，并出于效率的考虑进行预排序。默认情况下，缓冲区大小为 100 MB，此值可以通过改变 io.sort.mb 属性来调整。一旦缓冲内容达到阈值（io.sort.spill.percent，默认为 80%），一个后台线程便开始把内容写（spill）到磁盘中。在写磁盘的过程中，map 输出将继续被写到缓冲区，但如果在此期间缓冲区被填满，map 任务会阻塞知道写磁盘的过程完成。写入路径： mapred.local.dir。

分区

在写磁盘之前，线程首先根据数据最终要传送到的 reducer 把数据划分到相应的分区，在每个分区中，后台线程按键进行内排序，如果有一个 combiner,它会在排序后的输出上运行。

合并

一旦内存缓冲区达到溢出写对阈值，就会新建一个溢出写文件，因此在 map 任务完成最后一个记录输出之后，会有几个溢出写文件。在任务完成之前，溢出写文件被合并成一个已分区且已排序对输出文件。配置 io.sort.factor 控制一次合并多少个流，默认值为 10。

压缩

写磁盘压缩 map 输出，会让写磁盘的速度更快，节省磁盘空间，并且减少传给 reducer 的数据量。默认情况下，输出是不压缩的，但只要将mapred.compress.map.output 设置为 true，并使用 mapred.map.output.compression.codec 来指定压缩库。

数据传输

map 输出文件位于 map 任务的 tasktracker 的本地磁盘，reducer 通过 HTTP 方式得到 map 输出文件的分区。

reduce 端

reducer 一般执行数据的汇总和写出。copy ->merge

复制

reduce 任务需要集群上若干个 map 任务的输出作为其分区文件。每个 map 任务的完成时间不同，因此只要有一个 map 任务完成，reduce 任务就开始复制其输出，这就是 reduce 任务的复制阶段，reduce 任务有少量复制线程，默认值是 5 个线程，可以通过 mapred.reduce.paraller.copies 属性来指定，因此能并行得到 map 输出。如果 map 输出足够情况下，则会被复制到 reduce tasktracker 到内存（缓冲区大小mapred.job.shuffle.input.buffer.percent ）

reducer 如何知道从那个 tasktarcker 取得文件输出

map 任务完成后，它们会通知其父 tasktracker 状态已更新，tasktracker 通知 jobtracker 进程，通知通过心跳机制传输。reducer 中的一个线程定期询问 jobtracker，以便获得 map 的输出位置，知道它获得所有输出位置。

合并

每台机器上有个 TaskTracker 进程，管理机器上的资源，当资源空闲时，通知 JobTracker 根据分布 在该节点上启动 Map 进程或 Reduce 进程。

(3) 作业调优

调优项	调优方法
map 数	Map 数由 splitsize 决定，首先按 splitsize 切分数据，若剩余数据大小 < 1.1*splitsize ,则将剩余数据作为一个 split，splitsize 的取值方法：取 minSplit , maxSplit , blocksize 的中间值。
reduce 数	大部分场景下可由用户指定。
combiner	对于取最大值、词频统计场景，可以使用 combiner 对数据在 map 端进行 reduce 计算。
中间值的压缩
自定义序列
调整 shuffle

2. Spark

(1) spark运行流程

• 客户端执行 spark-submit 脚本，脚本内部会调用 spark-class 脚本，启动 SparkSubmit 类。
• SparkSubmit 类的内部会反射调用我们自己写的类的 main 方法。
• 此时开始创建 SparkContext。
• Driver 程序的代码运行到 action 操作，触发了 SparkContext 的 runJob 方法
• SparkContext 调用 DAGScheduler 的 runJob 函数，内部调用 DAGScheduler 的 submitJob 方法，返回一个 JobWaiter 对象。接着向 EventProcessLoop 的阻塞队列中 put 一个 JobSubmitted 事件。
• 这时候 DAGScheduler 的 onReceive 方法被调用，模式匹配，调用 handleJobSubmitted 方法，用来切分 stage。stage 的划分过程是递归调用，从前往后的划分 stage。
• 根据 final stage 递归找到第一个 stage，然后将第一个 stage 提交。
• 由于 stage 的类型不同，这里会有两种不同类型的 task，ShuffleMapTask 和 ResultTask。把 task 封装到 taskSet 里，把 Tasks 交给 TaskScheduler。（RPC 调用，向 Executor 提交 task）
• Executor 将 task 封装到 TaskRunner 对象中，将 TaskRunner 放入到 Executor 中的线程池中。
• 最后会调用 ShuffleMapTask 或 ResultTask 的 runTask 方法，执行业务逻辑。

Yarn-Cluster 模式

• Client 向 ResourceManager 提交任务。
• ResourceManager 根据用户任务分配合适的 Container,并选择一个 NodeManager 运行Container 节点启动一个 Spark AppMaster 进程。
• Spark App Master 启动后，向 ResourceManager 注册自己。
• 对于用户 Task, App Master 需要和 ResourceManager 协商获取用户 Task 所需的Container,成功后,Spark AppMaster 进程向
  发送任务给指定的 NodeManager 启动 Container,运行 Task。
  Spark-Submit Yarn-Cluster 提交流程源码分析

Yarn-Client 模式

• yarn-cluster 适用于生产环境；yarn-client 适用于交互、调试，希望立即看到app的输出。
• yarn-cluster 模式 Driver 在 AppMaster 中；yarn-client Driver 在 Client 中。

StandAlone Cluster 模式

• Client 向 Master 提交任务。
• Master 选择一个 Worker 节点启动 Driver 进程。
• Master 让其他的 Worker 启动 Executor 进程。
• Executor 启动后会向 Driver 中的 SchedulerBackend 进行注册。
• SchedulerBackend进程中包含 DAGScheduler ，它会根据用户程序，生成执行计划，并调度执行。
• 对于每个 stage 的 task ，都会被存放到 TaskScheduler 中，ExecutorBackend 向 SchedulerBackend 汇报的时候把 TaskScheduler 中的 task 调度到 ExecutorBackend 执行。

StandAlone Client 模式

• 客户端启动后直接运行 Driver 程序，启动 Driver 相关的工作：DAGScheduler和BlockManagerMaster等。

• 客户端的Driver向Master注册。

• Master还会让Worker启动Exeuctor。Worker创建一个ExecutorRunner线程，ExecutorRunner会启动ExecutorBackend进程。

• ExecutorBackend启动后会向Driver的SchedulerBackend注册。Driver的DAGScheduler解析作业并生成相应的Stage，每个Stage包含的Task通过TaskScheduler分配给Executor执行。

• 所有stage都完成后作业结束。

• Spark学习之路 （七）Spark 运行流程

• Spark架构与作业执行流程简介

(2) Sparkshuffle

Spark的Shuffle总体而言就包含两个基本的过程：Shuffle write和Shuffle read。shuffle write 有：sort，bypass，unsafe。

sort

步骤

• 使用 ExternalSorter 对 (partitionId,hash(key)) 元素进行排序，使得每个文件内数据局部有序。
• 使用堆进行 Merge Sort。

使用场景

• 需要 map 端聚合。

bypass

使用步骤

• 类似于 Hash，区别：根据用户指定的分区方法进行分区写入多个文件，最终将分区文件合并为 1 个。
• 同一个文件内数据按 hash(key) 排序。

使用场景

• 不支持 map 端聚合。

unsafe

步骤

使用场景

• 不支持 map 端聚合。
• 分区数小于 2^24。
• record 数小于 2^27。
• 可以使用堆外内存，降低磁盘 IO，直接进行数据传输。

Shuffle read 指的是 reducer 对属于自己的 FileSegment 文件进行fetch操作，这里采用的 netty 框架，效率明显好于 Mapreduce 的 http 传输。fetch操作会等到所有的Shuffle Write过程结束后再进行，这也是因为ShuffleMapTask可能并不在一个stage里面，需要在父stage执行之后提交才会进行子stage的执行。reducer通过fetch得到的FileSegment先放在缓冲区softBuffer中，默认大小48MB。Spark不要求Shuffle后的数据是全局有序的，所以没有必要等到shuffle read全部结束后再进行reduce，是可以并行处理的。

• Spark源码分析之Sort-Based Shuffle读写流程
• 彻底搞懂spark的shuffle过程（shuffle write)

(3) spark的内存管理机制

Spark中的内存管理(一)

(4) spark里面有哪些参数可以设置，有什么用

参数	功能
num-executors	Spark作业总共要用多少个Executor进程来执行。
executor-memory	设置每个Executor进程的内存。
executor-cores	设置每个Executor进程的CPU core数量。
driver-memory	设置Driver进程的内存。
spark.default.parallelism	该参数用于设置每个stage的默认task数量。
spark.shuffle.memoryFraction	设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的Executor内存的比例。
spark.storage.memoryFraction	设置RDD持久化数据在Executor内存中能占的比例。

Spark学习之路 （十二）SparkCore的调优之资源调优
spark submit参数及调优

(5) Spark 存储管理

(6) Spark算子

算子分类

Spark 算子大致可以分为以下两类：Transformation 算子和 Action 算子。
详细请看：Spark 算子分类

会产生 shuffle 的算子

cogroup、join、leftOutJoin、rightOutJoin、groupbykey和reducebykey 原理

join 流程

• 确认 join 完成后 RDD 的分区。
• 根据分区对进行 join 操作对 RDD 分别进行分区，写入文件中。
• 依次读取分区中数据，进行如下计算。

combineByKey 流程

• 根据分区策略在 Map 端将数据进行分区。
• 对数据进行排序、分组。
• 默认调用 mergeValue 函数对数据进行聚合。
• 将数据写入文件中。
• Reduce 端读取数据，调用 mergeCombiner 对数据进行聚合。
  combineByKey、reduceByKey、groupByKey、reduce、distinct

def combineByKey[C](
      createCombiner: V => C,
      mergeValue: (C, V) => C,
      mergeCombiners: (C, C) => C,
      partitioner: Partitioner,
      mapSideCombine: Boolean = true,
      serializer: Serializer = null): RDD[(K, C)] = self.withScope {
      combineByKeyWithClassTag(createCombiner, mergeValue, mergeCombiners,
      partitioner, mapSideCombine, serializer)(null)
  }

(7)Spark stage的切分、task资源分配、任务调度、master计算资源分配/(1)Spark Job、Task、Stage 区别与联系。

资源调度

应用程序如何获得资源。

任务调度

是通过 DAGScheduler、TaskScheduler、SchedulerBackend 等进行的作业调度

• Job : 根据 Action 划分。
• Stage : 根据窄依赖划分。
• Task : 根据分区个数确定每个 Stage 中的 Task 个数，根据 Partition 的位置确定 Task 的运行位置。
• DAG Schedule: 将 Job 分成 Stage 发送给 Task Schedule。
• Task Schedule: 将 Task 发送给 Executor。

(15)Spark数据倾斜解决方案

• 造成数据倾斜的原因
• 从数据来源分析：Kafka/HDFS、上一个 Stage

Spark性能优化之道——解决Spark数据倾斜（Data Skew）的N种姿势

3. Yarn

(1) Yarn 资源调度算法

• FIFO：不利于短任务。
• capacity scheduler：集群由很多队列组成，这些队列可能是层次结构的，每个队列有一个分配能力。在队列内部，作业根据 FIFO 进行调度。
• fair scheduler：目标是让每个用户公平地共享集群能力。如果只有一个作业运行，它会得到集群的所有资源。

Hadoop和Yarn中的任务调度算法和任务队列

(2) Yarn 组成架构

• ResourceManager:调度器和应用程序管理器。
• NodeManager：
• ApplicationMaster
• Container

（3）作业执行流程

• 应用程序向申请资源，启动 AppMaster。
• AppMaster 启动后，向 ResourceManager 注册。
• AppMaster 采用轮询的方式向 ResourceManager 申请资源。
• 申请到资源后，在对应的 NodeManager 启动 Container，并向 AppMaster 注册，运行 Task。

YARN基本框架和工作流程

4. SQL

(1) Hivesql/Spark SQL 是怎么转化为 MapReduce 任务的

1. SQL 语句经过 SqlParser 解析成 Unresolved LogicalPlan
2. 使用 Analyzer 结合数据数据字典(catalog)进行绑定，生成 resolved LogicalPlan
3. 使用 optimizer 对 resolved LogicalPlan 进行优化，生成 optimized LogicalPlan
4. 使用 SparkPlan 将 LogicalPlan 转换成 PhysicalPlan
5. 使用 prepareForExecution() 将 PhysicalPlan 转换成可执行物理计划
6. 使用 execute() 执行可执行物理计划
7. 生成 RDD。

(2) 数据倾斜

(3) Join 实现
ordinary join
hash join
sort merger join

大数据中Hadoop Hive发生数据倾斜的处理方案 绝对是干货

5. 流计算

Spark Streaming 和 Storm 的区别

处理模型,延迟

虽然这两个框架都提供可扩展性和容错性,它们根本的区别在于他们的处理模型。而 Storm 处理的是每次传入的一个事件，而 Spark Streaming 是处理某个时间段窗口内的事件流。因此，Storm 处理一个事件可以达到秒内的延迟，而 Spark Streaming 则有几秒钟的延迟。

容错、数据保证

在容错数据保证方面的权衡是，Spark Streaming 提供了更好的支持容错状态计算。在 Storm 中,每个单独的记录当它通过系统时必须被跟踪，所以 Storm 能够至少保证每个记录将被处理一次，但是在从错误中恢复过来时候允许出现重复记录。这意味着可变状态可能不正确地被更新两次。

另一方面，Spark Streaming 只需要在批级别进行跟踪处理，因此可以有效地保证每个 mini-batch 将完全被处理一次，即便一个节点发生故障。(实际上 Storm 的 Trident library 库也提供了完全一次处理。但是,它依赖于事务更新状态,这比较慢,通常必须由用户实现)。

总结

简而言之,如果你需要秒内的延迟，Storm 是一个不错的选择，而且没有数据丢失。如果你需要有状态的计算，而且要完全保证每个事件只被处理一次，Spark Streaming 则更好。Spark Streaming 编程逻辑也可能更容易，因为它类似于批处理程序(Hadoop)，特别是在你使用批次(尽管是很小的)时。

(29)spark streaming是怎么跟kafka交互的，具体代码怎么写的，程序执行流程是怎样的，这个过程中怎么确保数据不丢(直连和receiver方式)

通过 zookeeper 或其他外部存储获取 topic 每个 partition 的偏移量，消费完成后，将偏移量写入外部存储。

Kafka复制机制、分区多副本机制/

7. kafka partition 分区的策略是什么

消息发送到哪个分区上，有两种基本的策略，一是采用 Key Hash 算法，一是采用 Round Robin 算法。另外创建分区时，最好是 broker 数量的整数倍，这样才能是一个 Topic 的分区均匀的分布在整个 Kafka 集群中。

(30) kafka如何保证高吞吐的，kafka零拷贝，具体怎么做的。

• 分区：hash 分区或 轮询。
• 同步：ISR 中所有的副本返回 ACK，客户端提交提交成功。
• ISR 移除的条件：
  follower replica 落后于 leader 持续超过 replica.lag.max.messages。
  replica.lag.time.max.ms（replicas 响应 leader 的最长等待时间）如果一个 replica 失败导致发送拉取请求时间间隔超过replica.lag.time.max.ms，Kafka会认为此 replica 已经死亡会从同步副本列表从移除。

Kafka副本同步机制理解
Kafka介绍, kafka主题的副本机制
kafka producer自定义partitioner和consumer多线程

6. Zookeeper

(1)zookeeper怎么保证原子性，怎么实现分布式锁
(2)Zookeeper脑裂问题
(3)Zookeeper首领选取、节点类型、zookeeper实现原理
(4)hbase的ha，zookeeper在其中的作用

7. HDFS

(1) Hdfs读写流程，数据checkpoint流程
(2) hdfs的容错机制

8. Hbase

(1) Hbase 架构

1228818-20180402125236874-140096659.png

(2) Hbase 读写流程，数据 compact 流程
写

• Client 访问 Zookeeper,获取 Meta 表 Region 的位置。
• 根据 namespace、表名 和 rowkey 等信息，获得数据的 region 位置。
• 数据更新写入 Hlog。
• 根据列族确定写入 MemStore。
• 当 MemStore 的大小达到一定的阈值时，将 MemStore 刷到 StoreFile 中。
• 当 StoreFile 数量达到一定的阈值时，合并为一个 StoreFile。
• 当 StoreFile 达到一定的大小时，当前分割为多个 Region。

读

• Client 访问 Zookeeper，获取 Meta 表 Region 的位置。

• 根据 namespace、表名 和 rowkey 等信息，获得数据的 region 位置。

• 先从 BlockCache 中读数据，如果没有，从 MemStore 找数据，如果没有，再到StoreFile上读。

• HBase学习之路 （七）HBase 原理

9. ES

（1）ES 插入一条文档的过程。
（2）ES 集群查询一条文档过程。（query and fetch）

• 去 shard 节点根据 from、size 查询。
• 向 coordnitor 节点发送 id 和 相关性算分。
• coordinate 对相关性算分进行排序，然后再根据 size 取 id。
• 再根据 id 去 shard 中取数据。

10. 其他组件

10.1 Redis

(1) 数据类型、操作和实现

数据类型	操作	实现
String
List
Set
Map
ZSet

10.2 Kylin

使用预结算加速即时查询。

10.3 flume

解耦日志的产生和使用。