有道云笔记原文件和PDF文件已上传：https://download.csdn.net/download/z1941563559/12501011

1.HashMap和Hashtable区别
2.HashMap、Hashtable、ConcurrentHashMap的原理与区别
3.Java垃圾回收机制和生命周期
4.怎么解决Kafka数据丢失的问题
5.zookeeper是如何保证数据一致性的
6.hadoop和spark在处理数据时，处理出现内存溢出的方法有哪些？
7.Spark 如何调优
8.Flink和Spark的通信架构有什么异同
9.Java的代理
10.Java的内存溢出和内存泄露
11.Hadoop的组件有哪些？Yarn的调度器有哪些？
12.hadoop的shuffle过程
15.简述Spark集群运行的几种模式
16.Spark为什么比MapReduce快
17.spark工作机制
18.Spark的优化怎么做
19.数据本地性是在哪个环节确定的
20.RDD的弹性表现在哪几点
21.RDD有哪些缺陷
22.spark的数据本地性有哪几种
23.Spark为什么要持久化，一般什么场景下要进行persist操作
24.介绍一下join操作优化经验
25.描述Yarn执行一个任务的过程
26.Spark on yarn模式有哪些优点
27.谈谈你对container的理解
28.Spark使用parquet文件存储格式能带来什么好处
29.介绍partition和block有什么关联关系
30.Spark应用程序的执行过程是什么
31.不需要排序的hash shuffle是否一定比需要排序的sort shuffle速度快？
32.Sort-based shuffle的缺陷
33.spark.storage.memoryFraction参数的含义，实际生产中如何调优
34.介绍一下你对Unified Memory Management内存管理模型的理解
35.RDD中的reduceByKey与groupByKey哪个性能高
36.简述HBase的读写过程
37.在2.5亿个整数中，找出不重复的整数，注意：内存不足以容纳2.5亿个整数
38.CDH和HDP的区别
39.Java原子操作
40.Java封装、继承和多态
41.jvm模型
42.Flume taildirSorce重复读取数据解决方法
43.Flume如何保证数据不丢
44.Java 类加载过程
45.spark task运行原理
46.手写一个线程安全的单例
47.设计模式
48.impala 和 kudu 的适用场景，读写性能如何
49.Kafka ack原理
50.Flink TaskManager和Job Manager通信
51.Flink 双流 join方式
52.Flink state 管理和 checkpoint 的流程
53.Flink分层架构
54.Flink窗口
55.阐述Flink如何处理反压，相比Storm、Spark Streaming提供的反压机制，描述其实现有什么不同？
56.Spark Streaming 反压（Back Pressure）机制介绍
57.一文搞懂 Flink 的 Exactly Once 和 At Least Once
58.谈谈流计算中的『Exactly Once』特性
59.Time 深度解析
60.Flink Kafka 端到端Exactly-Once语义
61.Apache Flink 结合 Kafka 构建端到端的 Exactly-Once 处理
61.Flink Kafka Connector 与 Exactly-Once剖析

https://blog.csdn.net/huzechen/article/details/105321298
https://www.jianshu.com/p/b48f3ae30f23?utm_campaign=shakespeare

1.HashMap和Hashtable区别

HashMap和HashTable都实现了Map接口，但决定用哪一个之前先要弄清楚它们之间的分别。主要区别有：线程安全性，同步（synchronize），以及速度。
HashMap几乎可以等价于HashTable，除了HashMap是非synchronized的，并可以接收null（HashMap可以接受为null的键值，而HashTabel则不行）。
HashMap是非synchronized，而Hashtable是synchronized，这意味着Hashtable是线程安全的，多个线程可以共享一个Hashtable；而如果没有正确的同步的话，多个线程是不能共享HashMap的。Java 5提供了ConcurrentHashMap，它是HashTable的替代，比HashTable的扩展性更好。
另一个区别是HashMap的迭代器(Iterator)是fail-fast迭代器，而Hashtable的enumerator迭代器不是fail-fast的。所以当有其它线程改变了HashMap的结构（增加或者移除元素），将会抛出ConcurrentModificationException，但迭代器本身的remove()方法移除元素则不会抛出ConcurrentModificationException异常。但这并不是一个一定发生的行为，要看JVM。这条同样也是Enumeration和Iterator的区别。
由于Hashtable是线程安全的也是synchronized，所以在单线程环境下它比HashMap要慢。如果你不需要同步，只需要单一线程，那么使用HashMap性能要好过Hashtable。
HashMap不能保证随着时间的推移Map中的元素次序是不变的。
要注意的一些重要术语：
sychronized意味着在一次仅有一个线程能够更改Hashtable。就是说任何线程要更新Hashtable时要首先获得同步锁，其它线程要等到同步锁被释放之后才能再次获得同步锁更新Hashtable。
Fail-fast和iterator迭代器相关。如果某个集合对象创建了Iterator或者ListIterator，然后其它的线程试图“结构上”更改集合对象，将会抛出ConcurrentModificationException异常。但其它线程可以通过set()方法更改集合对象是允许的，因为这并没有从“结构上”更改集合。但是假如已经从结构上进行了更改，再调用set()方法，将会抛出IllegalArgumentException异常。
结构上的更改指的是删除或者插入一个元素，这样会影响到map的结构。
我们能否让HashMap同步？
HashMap可以通过下面的语句进行同步：
Map m = Collections.synchronizeMap(hashMap);
结论
Hashtable和HashMap有几个主要的不同：线程安全以及速度。仅在你需要完全的线程安全的时候使用Hashtable，而如果你使用Java 5或以上的话，请使用ConcurrentHashMap吧。

2.HashMap、Hashtable、ConcurrentHashMap的原理与区别

HashTable
底层数组+链表实现，无论key还是value都不能为null，线程安全，实现线程安全的方式是在修改数据时锁住整个HashTable，效率低，ConcurrentHashMap做了相关优化
初始size为11，扩容：newsize=oldsize2+1
计算index的方法：index=（hash & 0X7FFFFFFF）%tab.length
HashMap
底层数组+链表实现，可以存储null键和null值，线程不安全
初始size为16，扩容：newsize=oldsize2，size一定是2的n次幂
扩容针对整个Map，每次扩容时，原来数组中的元素一次重新计算存放位置，并重新插入
插入元素后才判断该不该扩容，有可能无效扩容（插入后如果扩容，如果没有再次插入，就会产生无效扩容）
当Map中元素总数超过Entry数组的75%，触发扩容操作，为了减少链表长度，元素分配更均匀
计算index方法：index=hash&（tab.length - 1）

HashMap的初始值还要考虑加载因子:
哈希冲突：若干Key的哈希值按数组大小取模后，如果落在同一个数组下标上，将组成一条Entry链，对Key的查找需要遍历Entry链上的每个元素执行equals()比较。
加载因子：为了降低哈希冲突的概率，默认当HashMap中的键值对达到数组大小的75%时，即会触发扩容。因此，如果预估容量是100，即需要设定100/0.75＝134的数组大小。
空间换时间：如果希望加快Key查找的时间，还可以进一步降低加载因子，加大初始大小，以降低哈希冲突的概率。

除此之外，hash表里还有一个“负载极限”，“负载极限”是一个0～1的数值，“负载极限”决定了hash表的最大填满程度。当hash表中的负载因子达到指定的“负载极限”时，hash表会自动成倍地增加容量（桶的数量），并将原有的对象重新分配，放入新的桶内，这称为rehashing。

HashMap和Hashtable的构造器允许指定一个负载极限，HashMap和Hashtable默认的“负载极限”为0.75，这表明当该hash表的3/4已经被填满时，hash表会发生rehashing。

“负载极限”的默认值（0.75）是时间和空间成本上的一种折中：
较高的“负载极限”可以降低hash表所占用的内存空间，但会增加查询数据的时间开销，而查询是最频繁的操作（HashMap的get()与put()方法都要用到查询）
较低的“负载极限”会提高查询数据的性能，但会增加hash表所占用的内存开销

ConcurrentHashMap
底层采用分段的数组+链表实现，线程安全
通过把整个Map分成N个Segment，可以提供相同的线程安全，但是效率提升N倍，默认提升16倍。（读操作不加锁，由于HashEntry的value变量是volatile[当一个共享变量被volatile修饰时，它会保证修改的值会立即被更新到主存，当有其他线程需要读取时，它会去内存中读取新值]的，也能保证读取到最新的值。）
Hashtable的synchronized是针对整张Hash表的，即每次锁住整张表让线程独占，ConcurrentHashMap允许多个修改操作并发进行，其关键在于使用了锁分离技术
有些方法需要跨段，比如size()和containsValue()，它们可能需要锁定整个表而不仅仅是某个段，这需要按顺序锁定所有段，操作完毕后，又按顺序释放所有段的锁
扩容：段内扩容（段内元素超过该段对应Entry数组长度的75%触发扩容，不会对整个Map进行扩容），插入前检测需不需要扩容，有效避免无效扩容

Hashtable和HashMap都实现了Map接口，但是Hashtable的实现是基于Dictionary抽象类的。Java5提供了ConcurrentHashMap，它是HashTable的替代，比HashTable的扩展性更好。

HashMap基于哈希思想，实现对数据的读写。当我们将键值对传递给put()方法时，它调用键对象的hashCode()方法来计算hashcode，然后找到bucket位置来存储值对象。当获取对象时，通过键对象的equals()方法找到正确的键值对，然后返回值对象。HashMap使用链表来解决碰撞问题，当发生碰撞时，对象将会储存在链表的下一个节点中。HashMap在每个链表节点中储存键值对对象。当两个不同的键对象的hashcode相同时，它们会储存在同一个bucket位置的链表中，可通过键对象的equals()方法来找到键值对。如果链表大小超过阈值（TREEIFY_THRESHOLD,8），链表就会被改造为树形结构。

在HashMap中，null可以作为键，这样的键只有一个，但可以有一个或多个键所对应的值为null。当get()方法返回null值时，即可以表示HashMap中没有该key，也可以表示该key所对应的value为null。因此，在HashMap中不能由get()方法来判断HashMap中是否存在某个key，应该用containsKey()方法来判断。而在Hashtable中，无论是key还是value都不能为null。

Hashtable是线程安全的，它的方法是同步的，可以直接用在多线程环境中。而HashMap则不是线程安全的，在多线程环境中，需要手动实现同步机制。

Hashtable与HashMap另一个区别是HashMap的迭代器（Iterator）是fail-fast迭代器，而Hashtable的enumerator迭代器不是fail-fast的。所以当有其它线程改变了HashMap的结构（增加或者移除元素），将会抛出ConcurrentModificationException，但迭代器本身的remove()方法移除元素则不会抛出ConcurrentModificationException异常。但这并不是一个一定发生的行为，要看JVM。

从类图中可以看出来在存储结构中ConcurrentHashMap比HashMap多出了一个Segment，而Segment是一个可重入锁。
ConcurrentHashMap是使用了锁分段技术来保证线程安全的。
锁分段技术：首先将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个数据段的时候，其他段的数据也能被其他线程访问。
ConcurrentHashMap提供了与HashTable和SynchronizedMap不同的锁机制。HashTable中采用的锁机制是一次锁住整个hash表，从而在同一时刻只能由一个线程对其进行操作；而ConcurrentHashMap中则是一次锁住一个桶。
ConcurrentHashMap默认将hash表分为16个桶，诸如get，put，remove等常用操作只锁住当前需要用到的桶。这样，原来只能一个线程进入，现在却能同时16个写线程执行，并发性能的提升是显而易见的。

3.Java垃圾回收机制和生命周期

c的垃圾回收是人工的，工作量大，但是可控性高。
java是自动化的，但是可控性很差，甚至有时会出现内存溢出的情况，内存溢出也就是jvm分配的内存中对象过多，超出了最大可分配内存的大小。
提到java的垃圾回收机制就不得不提一个方法：
System.gc()用于调用垃圾收集器，在调用时，垃圾收集器将运行以回收未使用的内存空间。它将尝试释放被丢弃对象占用的内存。
然后System.gc()调用附带一个免责申明，无法保证对垃圾收集器的调用。
所以System.gc()并不能说是完美主动进行了垃圾回收。

作为程序员还是很有必要了解一下gc，这也是面试过程中经常出现的一道题目。
我们从三个角度来理解gc：
1 jvm怎么确定哪些对象应该进行回收
2 jvm会在什么时候进行垃圾回收的动作
3 jvm到底是怎么清除垃圾对象的

1 jvm怎么确定哪些对象应该进行回收
对象是否会被回收两个经典算法：引用计数法，和可达性分析算法
引用计数法
简单的来说就是判断对象的引用数量。实现方式：给对象添加一个引用计数器，每当有引用对他进行引用时，计数器的值就加1，当引用失效，也就是不在执行此对象时，他的计数器的值就减1，若某一个对象的计数器的值为0，那么表示这个对象没有人对他进行引用，也就是意味着是一个失效的垃圾对象，就会被gc进行回收。

但是这种简单的算法在当前的jvm中并没有采用，原因是他并不能解决对象之间循环引用的问题。

假设有A和B两个对象之间互相引用，也就是说A对象中的一个属性是B，B中的一个属性时A,这种情况下由于他们的相互引用，从而是垃圾回收机制无法识别。

可达性分析算法
因为引用计数法的缺点又引入了可达性分析算法，通过判断对象的引用链是否可达来决定对象是否可以被回收。可达性分析算法是从离散数学中的图论引入的，程序把所有的引用关系看作一张图，通过一系列的名为GC Roots的对象作为起始点，从这些节点开始向下搜索，搜索所走过的路径称为引用链。当一个对象到 GC Roots 没有任何引用链相连（就是从 GC Roots 到这个对象不可达）时，则证明此对象是不可用的。

2 在确定了哪些对象可以被回收之后，jvm会在什么时候进行回收
1 会在cpu空闲的时候自动进行回收
2 在堆内存存储满了之后
3 主动调用System.gc()后尝试进行回收

3 如何回收
如何回收说的也就是垃圾收集的算法。
算法有四个：标记-清除算法，复制算法，标记-整理算法，分代收集算法

1 标记-清除算法
这里是最基础的一种算法，分为两个步骤，第一个步骤就是标记，也就是标记所有需要回收的对象，标记完成后就进行统一的回收掉那些带有标记的对象。这种算法优点是简单，缺点是效率问题，还有一个最大的缺点是空间问题，标记清除之后会产生大量不连续的内存碎片，当程序员在以后的运行过程中需要分配较大对象时无法找到足够的连续内存而造成内存空间浪费。

2 复制算法
复制将可用内存按容量划分为大小相等的两块，每次只使用其中一块。当这一块的内存用完了，就将还存活的对象复制到另外一块上面，然后再把已使用过的内存空间一次清理掉。这样使得每次都是对其中的一块进行内存回收，内存分配时也就不用考虑内存碎片等复杂情况。只是这种算法的代价是将内存缩小为原来的一半。
复制算法的执行过程如图：

复制收集算法在对象存活率较高时就要执行较多的复制操作，效率将会变低。更关键的是，浪费了一半的空间。

标记-整理算法：
标记整理算法与标记清除算法很相似，但最显著的区别是：标记清除算法仅对不存活的对象进行处理，剩余存活对象不做任何处理，造成内存碎片；而标记整理算法不仅对不存活对象进行处理清除，还对剩余的存活对象进行整理，最新整理，因此其不会产生内存碎片。

分代收集算法
分代收集算法是一种比较智能的算法，也是现在jvm使用最多的一种算法，他本身其实不是一个新的算法，而是他会在具体的场景自动选择以上三种算法进行垃圾回收。
那么现在的重点就是分代收集算法中说的自动根据场景进行选择。这个具体场景到底是什么场景。
场景其实指的是针对jvm的哪一个区域，1.7之前jvm把内存分为三个区域：新生代，老年代，永久代。

了解过场景之后再结合分代收集算法得出结论：1.在新生代中，每次垃圾收集时都发现有大批对象死去，只有少量存活，那就选用复制算法。只需要付出少量存活对象的复制成本就可以完成收集。2.老年代中因为对象存活率高、没有额外空间对他进行分配担保，就必须用标记-清除或者标记-整理。

每一次将对象放入JVM时，首先将创建的对象都放入到eden区域和其中一个survivor区域中；当eden区域和一个survivor区域放满了以后，这个时候会触发minor gc，把不再使用的对象全部清除，而剩余的对象放入另外一个servivor区域中。JVM中默认的eden，survivor1，survivor2的内存占比为8:1:1。当存活的对象在一个servivor中放不下的时候，就会将这些对象移动到老年代。如果JVM的内存不够大的话，就会频繁的触发minor gc，这样会导致一些短生命周期的对象进入到老年代，老年代的对象不断的囤积，最终触发full gc。一次full gc会使得所有其他程序暂停很长时间。

注意：
在jdk8的时候java废弃了永久代，但是并不意味着我们以上的结论失效，因为java提供了与永久代类似的叫做“元空间”的技术。
废弃用就嗲的原因：由于永久代内存经常不够用或发生内存泄露，爆出异常java.lang.OutOfMemoryError。元空间技术的本质和永久代类似。不过元空间与永久代之间最大的区别在于：元空间并不在虚拟机中，而是使用本地内存。也就是不局限与jvm可以使用系统的内存。理论上取决于32位/64位系统可虚拟的内存的小。

GC垃圾回收：
jvm按照对象的生命周期，将内存按“代”划分（将堆划分为多个地址池）：新生代、老年代和持久代（jdk1.8后移除持久代）

在jvm中程序（PC）计数器、Java栈、本地方法栈3个区域随线程而生、随线程而灭，因此这几个区域的内存分配和回收都具备确定性，就不需要过多考虑回收的问题，因为方法结束或者线程结束时，内存自然就跟随着回收了。而堆和方法区则不一样，这部分内存的分配和回收是动态的，正是垃圾回收器所需关注的部分。

java中新创建的对象会先被放在新生代区域，该区域对象使用频繁，jvm会在该区域使用不同算法回收一定的短期对象，如果某些对象使用次数达到一定限制后，那么该对象就会被放入老年代区域，老年代区域要比新生代区域更大一些（堆内存大部分分配给了老年代区域），而持久代保存的是类的元数据、常量、类静态变量等。

4.怎么解决Kafka数据丢失的问题

1）消费端弄丢了数据
唯一可能导致消费者弄丢数据的情况，就是说，你那个消费到了这个消息，然后消费者那边自动提交了offset，让kafka以为你已经消费好了这个消息，其实你刚刚准备处理这个消息，你还没处理，你自己就挂了，此时这条消息就丢了。

这不是一样么，大家都知道kafka会自动提交offset，那么只要关闭自动提交offset，在处理完之后自己手动提交offset，就可以保证数据不会丢。但是此时确实还是会重复消费，比如你刚处理完，还没提交offset，结果自己挂了，此时肯定会重复消费一次，自己保证幂等性就好了。

生产环境碰到的一个问题，就是说我们的kafka消费者消费到了数据之后是写到一个内存的queue里先缓冲一下，结果有的时候，你刚把消息写入内存queue，然后消费者会自动提交offset。

然后此时我们重启了系统，就会导致内存queue里还没来得及处理的数据就丢失了

2）kafka弄丢了数据
这块比较常见的一个场景，就是kafka某个broker宕机，然后重新选举partition的leader时。大家想想，要是此时其他的follower刚好还有些数据没有同步，结果此时leader挂了，然后选举某个follower成leader之后，他不就少了一些数据？这就丢了一些数据啊。

生产环境也遇到过，我们也是，之前kafka的leader机器宕机了，将follower切换成leader之后，就会发现说这个数据就丢了
所以此时一般是要求起码设置如下4个参数：

给这个topic设置replication.factor（用来设置主题的副本数）参数：这个值必须大于1，要求每个partition必须有至少2个副本

在kafka服务端设置min.insync.replicas参数：这个值必须大于1，这个是要求一个leader至少感知到有至少一个follower还跟自己保持联系，没掉队，这样才能确保leader挂了还有一个follower吧

在producer端设置acks=all：这个是要求每条数据，必须是写入所有replica之后，才能认为是写成功了

在producer端设置retries=MAX（很大很大很大的一个值，无限次重试的意思）：这个是要求一旦写入失败，就无限重试，卡在这里了

我们生产环境就是按照上述要求配置的，这样配置之后，至少在kafka broker端就可以保证在leader所在broker发生故障，进行leader切换时，数据不会丢失

3）生产者会不会弄丢数据
如果按照上述的思路设置了ack=all，一定不会丢失，要求是，你的leader接收到消息，所有的follower都同步到了消息之后，才认为本次写成功了。如果没满足这个条件，生产者会自动不断的重试，重试无限次。

5.zookeeper是如何保证数据一致性的

ZooKeeper是个集群，内部有多个server，每个server都可以连接多个client，每个client都可以修改server中的数据
ZooKeeper可以保证每个server内的数据完全一致，是如何实现的呢？
答：数据一致性是靠Paxos[pi ke sa si]算法保证的，Paxos可以说是分布式一致性算法的鼻祖，是ZooKeeper的基础
Paxos的基本思路：(深入解读zookeeper一致性原理)
假设有一个社团，其中有团员、*（决议小组成员）两个角色
团员可以向*申请提案来修改社团制度
*坐在一起，拿出自己收到的提案，对每个提案进行投票表决，超过半数通过即可生效
为了秩序，规定每个提案都有编号ID，按顺序自增
每个*都有一个社团制度笔记本，上面记着所有社团制度，和最近处理的提案编号，初始为0
投票通过的规则：
新提案ID 是否大于 *本中的ID，是 *举手赞同
如果举手人数大于*人数的半数，即让新提案生效
例如：
刚开始，每个*本子上的ID都为0，现在有一个*拿出一个提案：团费降为100元，这个提案的ID自增为1
每个*都和自己ID对比，一看 1>0，举手赞同，同时修改自己本中的ID为1
发出提案的*一看超过半数同意，就宣布：1号提案生效
然后所有*都修改自己笔记本中的团费为100元
以后任何一个团员咨询任何一个*：“团费是多少？”，*可以直接打开笔记本查看，并回答：团费为100元
可能会有极端的情况，就是多个*一起发出了提案，就是并发的情况
例如
刚开始，每个*本子上的编号都为0，现在有两个*（A和B）同时发出了提案，那么根据自增规则，这两个提案的编号都为1，但只会有一个被先处理
假设A的提案在B的上面，*们先处理A提案并通过了，这时，*们的本子上的ID已经变为了1，接下来处理B的提案，由于它的ID是1，不大于*本子上的ID，B提案就被拒绝了，B*需要重新发起提案
上面就是Paxos的基本思路，对照ZooKeeper，对应关系就是：
团员 -client
* -server
*的笔记本 -server中的数据
提案 -变更数据的请求
提案编号 -zxid（ZooKeeper Transaction Id）
提案生效 -执行变更数据的操作
ZooKeeper中还有一个leader的概念，就是把发起提案的权利收紧了，以前是每个*都可以发起提案，现在有了leader，大家就不要七嘴八舌了，先把提案都交给leader，由leader一个个发起提案
Paxos算法就是通过投票、全局编号机制，使同一时刻只有一个写操作被批准，同时并发的写操作要去争取选票，只有获得过半数选票的写操作才会被批准，所以永远只会有一个写操作得到批准，其他的写操作竞争失败只好再发起一轮投票
1）一致性保证
Zookeeper是一种高性能、可扩展的服务。Zookeeper的读写速度非常快，并且读的速度要比写的速度更快。另外，在进行读操作的时候，ZooKeeper依然能够为旧的数据提供服务。这些都是由于ZooKeepe所提供的一致性保证，它具有如下特点：
顺序一致性
客户端的更新顺序与它们被发送的顺序相一致。
原子性
更新操作要么成功要么失败，没有第三种结果。
单系统镜像
无论客户端连接到哪一个服务器，客户端将看到相同的ZooKeeper视图。
可靠性
一旦一个更新操作被应用，那么在客户端再次更新它之前，它的值将不会改变。。这个保证将会产生下面两种结果：
1．如果客户端成功地获得了正确的返回代码，那么说明更新已经成果。如果不能够获得返回代码（由于通信错误、超时等等），那么客户端将不知道更新操作是否生效。
2．当从故障恢复的时候，任何客户端能够看到的执行成功的更新操作将不会被回滚。
实时性
在特定的一段时间内，客户端看到的系统需要被保证是实时的（在十几秒的时间里）。在此时间段内，任何系统的改变将被客户端看到，或者被客户端侦测到。
给予这些一致性保证，ZooKeeper更高级功能的设计与实现将会变得非常容易，例如：leader选举、队列以及可撤销锁等机制的实现。
2）Leader选举
ZooKeeper需要在所有的服务（可以理解为服务器）中选举出一个Leader，然后让这个Leader来负责管理集群。此时，集群中的其它服务器则成为此Leader的Follower。并且，当Leader故障的时候，需要ZooKeeper能够快速地在Follower中选举出下一个Leader。这就是ZooKeeper的Leader机制，下面我们将简单介绍在ZooKeeper中，Leader选举（Leader Election）是如何实现的。
此操作实现的核心思想是：首先创建一个EPHEMERAL目录节点，例如“/election”。然后。每一个ZooKeeper服务器在此目录下创建一个SEQUENCE|EPHEMERAL 类型的节点，例如“/election/n_”。在SEQUENCE标志下，ZooKeeper将自动地为每一个ZooKeeper服务器分配一个比前一个分配的序号要大的序号。此时创建节点的ZooKeeper服务器中拥有最小序号编号的服务器将成为Leader。
在实际的操作中，还需要保障：当Leader服务器发生故障的时候，系统能够快速地选出下一个ZooKeeper服务器作为Leader。一个简单的解决方案是，让所有的follower监视leader所对应的节点。当Leader发生故障时，Leader所对应的临时节点将会自动地被删除，此操作将会触发所有监视Leader的服务器的watch。这样这些服务器将会收到Leader故障的消息，并进而进行下一次的Leader选举操作。但是，这种操作将会导致“从众效应”的发生，尤其当集群中服务器众多并且带宽延迟比较大的时候，此种情况更为明显。
在Zookeeper中，为了避免从众效应的发生，它是这样来实现的：每一个follower对follower集群中对应的比自己节点序号小一号的节点（也就是所有序号比自己小的节点中的序号最大的节点）设置一个watch。只有当follower所设置的watch被触发的时候，它才进行Leader选举操作，一般情况下它将成为集群中的下一个Leader。很明显，此Leader选举操作的速度是很快的。因为，每一次Leader选举几乎只涉及单个follower的操作。

6.hadoop和spark在处理数据时，处理出现内存溢出的方法有哪些？

1.map过程产生大量对象导致内存溢出
这种溢出的原因是单个map中产生了大量的对象导致的。
例如：rdd.map(x => for(i <- 1 to 10000)) yield i.toString，这个操作在rdd中，每个对象都产生了10000个对象，这肯定容易产生内存溢出的问题。针对这种问题，在不增加内存的情况下，可以通过减少每个Task的大小，以便达到每个Task即使产生大量的对象Executor的内存也能够装的下。具体做法可以在会产生大量对象的map操作之前调用repartition方法，分区成更小的快传入map。例如：rdd.repartition(10000).map(x => for(i <- 1 to 10000)) yield i.toString)。
面对这种问题注意，不能使用rdd.coalesce方法，这个方法只能减少分区，不能增加分区，不会有shuffle的过程。

2.数据不平衡导致的内存溢出
数据不平衡除了有可能导致内存溢出外，也有可能导致性能的问题，解决方法和上面说的类似，就是调用repartition重新分区。

3.coalesce[kou le si]调用导致内存溢出
因为hdfs中不适合存在小问题，所以spark计算后结果产生的文件太小，我们会调用coalesce合并文件再存入hdfs中。但是这会导致一个问题，例如在coalesce之前有100个文件，这也意味着能够有100个task，现在调用coalesce(10)，最后只产生10个文件，因为coalesce并不是shuffle操作，这意味着coalesce并不是按照我原本想的那样先执行100个task，在将task的执行结果合并成10个，而是从头到尾只有10个task在执行，原本100个文件是分开执行的，现在每个task同时一次读取10个文件，使用的内存时原来的10倍，这导致了OOM。解决这个问题的方法是令程序按照我们想的100个task再将结果合并成10个文件，这个问题同样可以通过repartition解决，调用repartition(10)，因为这就有一个shuffle过程，shuffle前后是两个stage，一个100个分区，一个10个分区，就能按照我们的想法执行。

4.shuffle后内存溢出
shuffle内存溢出的情况可以说都是shuffle后，单个文件过大导致的。在Spark中，join，reduceByKey这一类型的过程，都会有shuffle的过程，在shuffle的使用，需要传入一个partitioner，大部分Spark中的shuffle操作，默认的partitioner都是HashPartitioner，默认值是父RDD中最大的分区数，这个参数通过spark.default.parallelism控制（在spark-sql中用spark.sql.shuffle.partitions），spark.default.parallelism参数只对HashPartitioner有效，所以如果是别的Partitioner或者自己实现的Partitioner就不能使用spark.default.parallelism这个参数来控制shuffle的并发量了。如果是别的Partitioner导致的shuffle内存溢出，就要从Partitioner的代码增加partitions的数量。

5.standalone模式下资源分配不均匀导致内存溢出
在standalone的模式下如果配置了-total-executor-cores和-executor-memory这两个参数，但是没有配置-executor-cores这个参数的话，就有可能导致，每个executor的memory是一样的，但是cores的数量不同，那么在cores数量多的executor中，由于能够同时执行多个task，就容易导致内存溢出的情况。这种情况的解决方法就是同时配置-executor-cores或者spark-executor-cores参数，确保executor资源分配均匀。

6.在RDD中，共用对象能够减少OOM的情况
这个比较特殊，这里说记录一下，遇到过一种情况，类似这样rdd.flatMap(x=>for(i <- 1 to 1000) yield (“key”,”value”))导致OOM，但是在同样的情况下，使用rdd.flatMap(x=>for(i <- 1 to 1000) yield “key”+”value”)就不会有OOM的问题，这是因为每次(“key”,”value”)都产生一个Tuple对象，而”key”+”value”，不管多少个，都只有一个对象，指向常量池。具体测试如下：

这个例子说明(“key”,”value”)和(“key”,”value”)在内存中是存在不同位置的,也就是存了两份,但是”key”+”value”虽然出现了两次,但是只存了一份,在同一个地址,这用到了JVM常量池的知识.于是乎,如果RDD中有大量的重复数据,或者Array中需要存大量重复数据的时候我们都可以将重复数据转化为String,能够有效的减少内存使用.

7.Spark 如何调优

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

2）设置num-executors参数
参数说明：该参数用于设置Spark作业总共要用多少个Executor进程来执行。Driver在向YARN集群管理器申请资源时，YARN集群管理器会尽可能按照你的设置来集群的各个工作节点上，启动相应数量的Executor进程。这个参数非常之重要，如果不设置的话，默认只会给你启动少量的Executor进程，此时你的Spark作业的运行速度是非常慢的。

参数调优建议：该参数设置的太少，无法充分利用集群资源；设置的太多的话，大部分队列可能无法给予充分的资源。针对数据交换的业务场景，建议该参数设置1-5。

3）设置executor-memory参数
参数说明：该参数用于设置每个executor进程的内存。executor内存的大小，很多时候直接决定了Spark作业的性能，而且跟常见的JVM OOM异常也有直接关联。

参数调优建议：针对数据交换的业务场景，建议本参数设置在512M及以下

4）executor-cores
参数说明：该参数用于设置每个executor进程的cpu core数量 。这个参数决定了每个executor进程并行执行task线程的能力。因为每个CPU core同时只能执行一个task线程，因此每个executor进程的cpu core数量越多，越能够快速地执行完分配给自己的所有task线程。

参数调优建议：executor的cpu core数量设置为2~4个较为合适。建议，如果是跟他人共享一个队列，那么num-executors * executor-cores不要超过队列总cpu core的1/3 ~1/2左右比较合适，避免影响其他人的作业运行。

5）driver-memory
参数说明：该参数用于设置Driver进程的内存。

参数调优建议：driver的内存通常来说不设置，或者设置512M以下就够了。唯一需要注意的一点是，如果需要使用collect算子将RDD的数据全部拉取到Driver上进行处理，那么必须确保Driver的内存足够大，否则会出现OOM内存溢出的问题。

6）spark.default.parallelism
参数说明：该参数用于设置每个stage的默认task数量。这个参数极为重要，如果不设置可能会直接影响你的spark作用性能。

参数调优建议：如果不设置这个参数，Spark自己会根据底层HDFS的block数量来说设置task的数量，默认是一个hdfs block对应一个task。Spark官网建议的设置原则是，设置该参数为num-executors * executor-cores的2~3倍较为合适，此时可以充分地利用Spark集群的资源。针对数据交换的场景，建议此参数设置为1-10。

7）spark.storage.memoryFraction
参数说明：该参数用于设置RDD持久化数据在Executor内存中占的比例，默认是0.6。也就是说，默认Executor 60%的内存，可以用来保存持久化的RDD数据。根据你选择的不同的持久化策略，如果内存不够时，可能数据就不会持久化，或者数据会写入磁盘。

参数调优建议：如果Spark作业中，有较多的RDD持久化操作，该参数的值可以适当提高一些，保证持久化的数据能够容纳在内存中。避免内存不否缓存所有的数据，导致数据只能写入磁盘中，降低了性能。但是如果spark作业中的shuffle类操作比较多，而持久化操作比较少，那么这个参数的值适当降低一些比较合适。如果发现作业由于频繁的gs导致运行缓慢（通过spark web ui可以观察到作业的gc耗时），意味着task执行用户代码的内存不够用，那么同样调低这个参数的值。针对数据交换的场景，建议降低此参数值到0.2-0.4。

8）spark.shuffle.memoryFraction
参数说明：该参数用于设置shuffle过程中一个task拉取到上个stage的task的输出后，进行聚合操作时能够使用的executor内存的比例，默认是0.2。也就是说，executor默认只有20%的内存用来进行该操作。shuffle操作在进行聚合时，如果发现使用的内存超出了这个20%的限制，那么多余的数据就会溢写到磁盘文件中去，此时就会极大地降低性能。

参数调优建议：如果Spark作业中的RDD持久化操作较少，shuffle操作较多时，建议降低持久化操作的内存占比，提高shuffle操作的内存占比比例，避免shuffle过程中数据过多时内存不够用，必须溢写到磁盘上，降低了性能。如果发现作业由于频繁的gc导致运行缓慢，意味着task执行用户代码的内存不够用，那么同样建议调低这个参数的值。针对数据交换的场景，建议此值为0.1或以下。

./bin/spark-submit
–master yarn-cluster
–num-executors 1
–executor-memory 512M
–executor-cores 2
–driver-memory 512M
–conf spark.default.parallelism=2
–conf spark.storage.memoryFraction=0.2
–conf spark.shuffle.memoryFraction=0.1 \

8.Flink和Spark的通信架构有什么异同

首先它们有哪些共同点？flink和spark都是Apache软件基金会（ASF）旗下*项目，都是通用数据处理平台。它们可以应用在很多的大数据应用和处理环境。并且有如下扩展：

并且两者均可在不依赖与其他环境的情况下运行与standalone模式，或是运行在基于hadoop（YARN，HDFS）之上，由于它们均是运行于内存，所以他们表现的都比hadoop要好很多。

然而它们在实现上还是有很多不同点：
在spark 1.5.x之前的版本，数据集的大小不能大于机器的内存数。

Flink在进行集合的迭代转换时可以是循环或是迭代计算处理。这使得Join算法、对分区的链接和重用以及排序可以选择最优算法。当然flink也是一个很强大的批处理工具。flink的流式处理的是真正的流处理。流式数据一旦进入就实时处理，这就允许流数据灵活地在操作窗口。它甚至可以在使用水印的流数中处理数据。此外，flink的代码执行引擎还对现有使用storm，MapReduce等有很强的兼容性。

Spark在另一方面是基于弹性分布式数据集（RDD），这给予spark给予内存数据结构的函数式编程。它可以通过固定的内存基于大批量的计算。spark Streaming把流式数据封装成小的批处理，也就是它收集在一段时间内到达的所有数据，并在收集的数据上运行一个常规批处理程序。同时一边收集下一个小的批处理数据。

9.Java的代理

代理模式是什么？
代理模式是一种设计模式，简单说即是在不改变源码的情况下，实现对目标对象的功能扩展。

10.Java的内存溢出和内存泄露

内存溢出：out of memory，是指程序在申请内存时，没有足够的内存空间供其使用，出现out of memory；比如申请了一个integer，但给它存了long才能存下的数，那就是内存溢出。
内存泄露：memory leak，是指程序在申请内存后，无法释放已申请的内存空间。

11.Hadoop的组件有哪些？Yarn的调度器有哪些？

Hadoop常用组件介绍
1.Hive
Hive是将Hadoop包装成使用简单的软件，用户可以用比较熟悉的SQL语言来调取数据，也就是说，Hive其实就是将Hadoop包装成MySQL。Hive适合使用在对实时性要求不高的结构化数据处理。像是每天、每周用户的登录次数、登录时间统计；每周用户增长比例之类的BI应用。

2.HBase
HBase是用来存储和查询非结构化和半结构化数据的工具，利用row key的方式来访问数据。HBase适合处理大量的非结构化数据，例如图片、音频、视频等，在训练机器学习时，可以快速的透过标签将相对应的数据全部调出。

3.Storm
前面两个都是用来处理非实时的数据，对于某些讲求实时性（毫秒级）的应用，就需要使用Storm。Storm也是具有容错和分布式计算的特性，架构为master-slave，可横向扩充多节点进行处理，每个节点每秒可以处理上百万条记录。可用在金融领域的风控上。

4.Impala
Impala和Hive的相似度很高，最大的不同是Impala使用了基于MPP的SQL查询，实时性比MapReduce好很多，但是无法像Hive一样可以处理大量的数据。Impala提供了快速轻量查询的功能，方便开发人员快速的查询新产生的数据。

YARN提供了三种调度策略
1.FIFO-先进先出调度器
YARN默认情况下使用的是该调度器，即所有的应用程序都是按照提交的顺序来执行的，这些应用程序都放在一个队列中，只有在前面的一个任务执行完成之后，才可以执行后面的任务，依次执行

缺点：如果有某个任务执行时间较长的话，后面的任务都要处于等待状态，这样的话会造成资源的使用率不高；如果是多人共享集群资源的话，缺点更是明显

2.capacity-scheduler-容量调度器 [ke pai ce ti]
针对多用户的调度，容量调度器采用的方法稍有不同。集群由很多队列组成（类似于任务池），这些队列可能是层次结构的（因此，一个队列可能是另一个队列的子队列），每个队列被分配有一定的容量。这一点与公平调度器类似，只不过在每个队列的内部，作业根据FIFO的方式（考虑优先级）调度。本质上，容量调度器允许用户或组织（使用队列自行定义）为每个用户或组织模拟出一个使用FIFO调度策略的独立MapReduce集群。相比之下，公平调度器（实际上也支持作业池内的FIFO调度，使其类似于容量调度器）强制池内公平共享，使运行的作业共享池内的资源。

3.Fair-scheduler-公平调度器
所谓的公平调度器指的是，旨在让每个用户公平的共享集群的能力。如果是只有一个作业在运行的话，就会得到集群中所有的资源。随着提交的作业越来越多，限制的任务槽会以“让每个用户公平共享集群”这种方式进行分配。某个用户的耗时短的作业将在合理的时间内完成，即便另一个用户的长时间作业正在运行而且还在运行过程中。
作业都是放在作业池中的，默认情况下，每个用户都有自己的作业池。提交作业数较多的用户，不会因此而获得更多的集群资源。可以用map和reduce的任务槽数来定制作业池的最小容量，也可以设置每个池的权重。
公平调度器支持抢占机制。所以，如果一个吃在特定的一段时间内未能公平的共享资源，就会终止运行池中得到过多的资源的任务，把空出来的任务槽让给运行资源不足的作业池。

12.hadoop的shuffle过程

以wordcount为例，假设有5个map和3个reduce：

map阶段

1、在map task执行时，它的输入数据来源于HDFS的block，当然在MapReduce概念中，map task只读取split。Split与block的对应关系可能是多对一，默认是一对一。

2、在经过mapper的运行后，我们得知mapper的输出是这样一个key/value对： key是“hello”， value是数值1。因为当前map端只做加1的操作，在reduce task里才去合并结果集。这个job有3个reduce task，到底当前的“hello”应该交由哪个reduce去做呢，是需要现在决定的。

分区（partition）

MapReduce提供Partitioner接口，它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力，如果用户自己对Partitioner有需求，可以订制并设置到job上。

一个split被分成了3个partition。

排序sort

在spill写入之前，会先进行二次排序，**首先根据数据所属的partition进行排序，然后每个partition中的数据再按key来排序。**partition的目是将记录划分到不同的Reducer上去，以期望能够达到负载均衡，以后的Reducer就会根据partition来读取自己对应的数据。接着运行combiner(如果设置了的话)，combiner的本质也是一个Reducer，其目的是对将要写入到磁盘上的文件先进行一次处理，这样，写入到磁盘的数据量就会减少。

溢写（spill）

Map端会处理输入数据并产生中间结果，这个中间结果会写到本地磁盘，而不是HDFS。每个Map的输出会先写到内存缓冲区中， 缓冲区的作用是批量收集map结果，减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前，key与value值都会被序列化成字节数组。 当写入的数据达到设定的阈值时，系统将会启动一个线程将缓冲区的数据写到磁盘，这个过程叫做spill。

这个溢写是由单独线程来完成，不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出，所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8。

将数据写到本地磁盘产生spill文件(spill文件保存在{mapred.local.dir}指定的目录中，MapReduce任务结束后就会被删除)。

合并（merge）

每个Map任务可能产生多个spill文件，在每个Map任务完成前，会通过多路归并算法将这些spill文件归并成一个文件。这个操作就叫merge(spill文件保存在{mapred.local.dir}指定的目录中，Map任务结束后就会被删除)。一个map最终会溢写一个文件。

至此，Map的shuffle过程就结束了。

Reduce阶段

Reduce端的shuffle主要包括三个阶段，copy、sort(merge)和reduce。

copy

首先要将Map端产生的输出文件拷贝到Reduce端，但每个Reducer如何知道自己应该处理哪些数据呢？因为Map端进行partition的时候，实际上就相当于指定了每个Reducer要处理的数据(partition就对应了Reducer)，**所以Reducer在拷贝数据的时候只需拷贝与自己对应的partition中的数据即可。**每个Reducer会处理一个或者多个partition，但需要先将自己对应的partition中的数据从每个Map的输出结果中拷贝过来。

merge

Copy过来的数据会先放入内存缓冲区中，这里的缓冲区大小要比map端的更为灵活，它基于JVM的heap size设置，因为Shuffle阶段Reducer不运行，所以应该把绝大部分的内存都给Shuffle用。

这里需要强调的是：

merge阶段，也称为sort阶段，因为这个阶段的主要工作是执行了归并排序。从Map端拷贝到Reduce端的数据都是有序的，所以很适合归并排序。

merge有三种形式：1)内存到内存 2)内存到磁盘 3)磁盘到磁盘。默认情况下第一种形式不启用，让人比较困惑，是吧。

当copy到内存中的数据量到达一定阈值，就启动内存到磁盘的merge，即第二种merge方式，与map 端类似，这也是溢写的过程，这个过程中如果你设置有Combiner，也是会启用的，然后在磁盘中生成了众多的溢写文件。这种merge方式一直在运行，直到没有map端的数据时才结束。

然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。

reduce

不断地merge后，最后会生成一个“最终文件”。为什么加引号？因为这个文件可能存在于磁盘上，也可能存在于内存中。对我们来说，当然希望它存放于内存中，直接作为Reducer的输入，但默认情况下，这个文件是存放于磁盘中的。至于怎样才能让这个文件出现在内存中，参见性能优化篇。

然后就是Reducer执行，在这个过程中产生了最终的输出结果，并将其写到HDFS上。

15.简述Spark集群运行的几种模式

a.local本地模式
spark不一定非要跑在hadoop集群，可以在本地，起多个线程的方式来指定。将Spark应用程序的方式直接运行在本地，一般都是为了方便调试，本地模式分三类
local：只启动一个executor
local[k]：启动k个executor
local[*]：启动跟CPU数目相同的executor

b.spark内置standalone集群模式
Spark自带的一个简单的集群管理器，这使得启动一个Spark集群变得非常简单。
分布式部署集群，自带完整的服务，资源管理和任务监控是Spark自己监控，这个模式也是其他模式的基础。

c.yarn集群模式
分布式部署集群，资源和任务监控交给yarn管理，但是目前仅支持粗粒度资源分配方式，包括cluster和client运行模式，cluster适合生产，driver运行在集群子节点，具有容错功能，client适合调试，driver运行在客户端。

大数据Spark面试题（一） - 蓦然的文章 - 知乎
https://zhuanlan.zhihu.com/p/107354908

16.Spark为什么比MapReduce快

1）基于内存计算，减少低效的磁盘交互
2）高效的调度算法，基于DAG
3）容错机制Lineage，精华部分就是DAG的Lineage[li ni a j]

17.spark工作机制

①构建Application的运行环境，Driver创建一个SparkContext

②SparkContext向资源管理器（standalone、mesos、yarn）申请Executor资源，资源管理器启动StandaloneExecutorbackend（Executor）
③Executor向SparkContext申请Task
④SparkContext将应用程序分发给Executor
⑤SparkContext构建DAG图，DAGScheduler将DAG图解析成Stage，每个stage有多个task，形成taskset发送给task scheduler，由task scheduler将task发送给Executor运行
⑥Task在executor上运行，运行完释放所有资源

18.Spark的优化怎么做

spark调优比较复杂，但是大体可以分为三个方面来进行
1）平台层面的调优：防止不必要的jar包分发，提高数据本地性，选择高校的存储格式如parquet
2）应用程序层面的调优：过滤操作符的优化降低过多小任务，降低单条记录的资源开销，处理数据倾斜，复用RDD进行缓存，作业并行化执行等等
3）JVM层面的调优：设置合适的资源量，设置合理的JVM，启用高校的序列化方法比如kyro，增大off heap内存等等

19.数据本地性是在哪个环节确定的

具体的task运行在哪台机器上，dag划分stage的时候确定的

20.RDD的弹性表现在哪几点

容错 缓存
1）自动的进行内存和磁盘存储切换
2）基于Lineage的高效容错
3）task如果失败会自动进行特定次数的重试
4）stage如果失败会自动进行特定次数的重试，而且只会计算失败的分片
5）checkpoint的persist，数据计算之后持久化缓存
6）数据调度弹性，DAG TASK调度和资源无关
7）数据分片的高度弹性

21.RDD有哪些缺陷

1）不支持细粒度的写和更新操作，spark写数据是粗粒度的。所谓粗粒度，就是批量写入数据，为了提高效率。但是读数据是细粒度的也就是说可以一条条的读。
2）不支持增量迭代计算，Flink支持

22.spark的数据本地性有哪几种

Spark中的数据本地性有三种：
1）PROCESS_LOCAL是指读取缓存在本地节点的数据
2）NODE_LOCAL是指读取本地节点硬盘数据
3）ANY是指读取非本地节点数据

通常读取数据PROCESS_LOCAL > NODE_LOCAL > ANY，尽量使数据以PROCESS_LOCAL或NODE_LOCAL方式读取。其中PROCESS_LOCAL还和cache有关，如果rdd经常用的话该rdd cache到内存中，注意，由于cache是lazy的，所以必须通过一个action的触发，才能真正的将该rdd cache到内存中。

23.Spark为什么要持久化，一般什么场景下要进行persist操作

为什么要持久化？
spark所有复杂一点的算法都会有persist身影，spark默认数据放在内存，spark很多内容都是放在内存的，非常适合高速迭代，1000个步骤只有第一个输入数据，中间不产生临时数据，但分布式系统风险很高，所以容易出错，就要容错，rdd出错或者分片可以根据血统算出来，如果没有对父rdd进行persist或者cache的话，就需要从头做。

以下场景会使用persist
1）某个步骤计算非常耗时，需要进行persist持久化
2）计算链条非常长，重新恢复要算很多步骤，很好使，persist
3）checkpoint所在的rdd要持久化persist。checkpoint前要持久化，写个rdd.cache或者rdd.persist，将结果保存起来，再写checkpoint操作，这样执行起来会非常快，不需要重新计算rdd链条了。checkpoint之前一定会进行persist。
4）shuffle之后要persist，shuffle要进行网络传输，风险很大，数据丢失重来，恢复代价很大
5）shuffle之前进行persist，框架默认将数据持久化到磁盘，这个是框架自己做的

24.介绍一下join操作优化经验

join其实常见的就分为两类：map-side join和reduce-side join。当大表和小表join时，用map-side join能显著提高效率。将多份数据进行关联是数据处理过程中非常普遍的用法，不过在分布式计算系统中，这个问题往往会变的非常麻烦，因为框架提供的join操作一般会将所有数据根据key发送到所有的reduce分区中去，也就是shuffle的过程。造成大量的网络以及磁盘IO消耗，运行效率极其低下，这个过程一般被称为reduce-side join。如果其中有张表较小的话，我们则可以自己实现在map端实现数据关联，跳过大量数据进行shuffle的过程，运行时间得到大量缩短，根据不同数据可能会有几倍到数十倍的性能提升。

25.描述Yarn执行一个任务的过程

1）客户端client向ResourceManager提交Application，ResourceManager接收Application并根据集群资源状况选取一个node来启动Application的任务调度器driver（ApplicationMaster）
2）ResourceManager找到那个node，命令该node上的NodeManager来启动一个新的JVM进程运行程序的driver（ApplicationMaster），driver（ApplicationMaster）启动时会首先向ResourceManager注册，说明由自己来负责当前程序的运行
3）driver（ApplicationMaster）开始下载相关jar包等各种资源，基于下载的jar等信息决定向ResourceManager申请具体的资源内容
4）ResourceManager接收到driver（ApplicationMaster）提出的申请后，会最大化的满足资源分配请求，并发送资源的元数据信息给driver（ApplicationMaster）
5）driver（ApplicationMaster）收到发过来的资源元数据信息后根据元数据信息发质量给具体的机器上的NodeManager，让其启动具体的container
6）NodeManager收到driver发来的指令，启动container，container启动后必须向driver（ApplicationMaster）注册
7）driver（ApplicationMaster）收到container的注册，开始进行任务的调度和计算，直到任务完成

注意：如果ResourceManager第一次没有能够满足driver（APplicationMaster）的资源请求，后续发现有空闲的资源，会主动向的river（ApplicationMaster）发送可用资源的元数据信息以提供更多的资源用于当前程序的运行。

26.Spark on yarn模式有哪些优点

1）与其他计算框架共享集群资源（Spark框架与MapReduce框架同时运行，如果不用yarn进行资源分配，MapReduce分到的内存资源会很少，效率低下）；资源按需分配，进而提高集群资源利用等
2）相较于Spark自带的standalone模式，Yarn的资源分配更加细致
3）Application部署简化，例如Spark，Storm等多种框架的应用由客户端提交后，由yarn负责资源的管理和调度，利用container作为资源隔离的单位，以它为单位去使用内存，cpu等
4）yarn通过队列的方式，管理同时运行在yarn集群中的多个服务，可根据不同类型的应用程序负载情况，调整对应的资源使用量，实现资源弹性管理

27.谈谈你对container的理解

1）Container作为资源分配和调度的单位，其中封装了的资源如内存，CPU，磁盘，网络带宽等。目前yarn仅仅封装内存和CPU
2）Container由ApplicationMaster向ResourceManager申请的，由ResourceManager中的资源调度器异步分配给ApplicationMaster
3）Container的运行是由ApplicationMaster向资源所在的NodeManager发起的，Container运行时需要提供内部执行的任务命令

28.Spark使用parquet文件存储格式能带来什么好处

1）如果说HDFS是大数据时代分布式文件系统首选标准，那么parquet则是整个大数据时代文件存储格式实时首选标准
2）速度更快：从使用spark sql操作普通文件CSV和parquet速度对比上，绝大多数情况会比使用CSV等普通文件速度提升10倍左右，在一些普通文件系统无法在spark上成功运行的情况下，使用parquet很多时候可以成功运行
3）parquet的压缩技术非常稳定出色，在spark sql中对压缩技术的处理可能无法正常的完成工作（例如会导致lost task，lost executor）但是此时如果使用parquet就可以正常的完成
4）极大的减少磁盘I/O，通常情况下能够减少75%的存储空间，由此可以极大的减少spark sql处理数据的时候的数据输入内容，尤其是在spark 1.6.x中有个下推过滤器在一些情况下可以极大的减少磁盘的IO和内存的占用
5）spark 1.6.x parquet方式极大的提升了扫描的吞吐量，极大提高了数据的查找速度spark1.6和spark1.5.x相比而言，提升了大约一倍的速度，在spark1.6.x中，操作parquet的时候CPU也进行了极大的优化，有效的降低了cpu消耗
6）采用parquet可以极大的优化spark的调度和执行。我们测试spark如果用parquet可以有效减少stage的执行消耗，同时可以优化执行路径

29.介绍partition和block有什么关联关系

1）hdfs中的block是分布式存储的最小单元，等分，可设置冗余，这样设计有一部分磁盘空间的浪费，但是整齐的block大小，便于快速找到、读取对应的内容
2）spark中的partition是弹性分布式数据集RDD的最小单元，RDD是由分布在各个节点上的partition组成的。partition是指的spark在计算过程中，生成数据在计算空间内最小单元，同一份数据（RDD）的partition大小不一，数据不定，是根据application里的算子和最初读入的数据分块数量决定
3）block位于存储空间、partition位于计算空间，block的大小是固定的、partition大小是不固定的，是从2个不同的角度去看数据

30.Spark应用程序的执行过程是什么

1）构建Spark Application的运行环境（启动SparkContext），SparkContext向资源管理器（可以是standalone、Mesos或yarn）注册并申请运行Executor资源；
2）资源管理器分配Executor资源并 启动StandaloneExecutorBackend，Executor运行情况将随着心跳发送到资源管理器上
3）SparkContext构建成DAG图，将DAG图分解成Stage，并把TaskSet发送给Task Scheduler。Executor向SparkContext申请Task，Task Scheduler将Task发放给Executor运行同时SparkContext将应用程序代码发放给Executor
4）Task在Executor上运行，运行完毕释放所有资源

31.不需要排序的hash shuffle是否一定比需要排序的sort shuffle速度快？

不一定，当数据规模小，Hash shuffle快于Sorted Shuffle数据规模大的时候；当数据量大，sorted Shuffle会比Hash shuffle快很多，因为数量大的有很多小文件，不均匀，甚至出现数据倾斜，消耗内存大，1.x之前spark使用hash，适合处理中小规模，1.x之后，增加了Sorted shuffle，Spark更能胜任大规模处理了。

32.Sort-based shuffle的缺陷

1）如果mapper中task的数量过大，依旧会产生很多小文件，此时在shuffle传递数据的过程中reduce端，reduce会需要同时大量的记录进行反序列化，导致大量的内存消耗和GC的巨大负担，造成系统缓慢甚至崩溃
2）如果需要在分片内也进行排序，此时需要进行mapper端和reducer端的两次排序

33.spark.storage.memoryFraction参数的含义，实际生产中如何调优

1）用于设置RDD持久化数据在Executor内存中能占的比例，默认是0.6，默认是Executor 60%的内存，可以用来保存持久化的RDD数据。根据你选择的不同的持久化策略，如果内存不够时，可能数据就不会持久化，或者数据会写入磁盘
2）如果持久化操作比较多，可以提高spark.storage.memoryFraction参数，使得更多的持久化数据保存在内存中，提高数据的读取性能，如果shuffle的操作比较多，有很多的数据读写操作到jvm中，那么应该调小一点，节约出更多的内存给JVM，避免过多的JVM gs发生。在web ui中观察如果发现gc时间很长，可以设置spark.storage.memoryFraction更小一点。

34.介绍一下你对Unified Memory Management内存管理模型的理解

Spark中的内存使用分两部分：执行（execution）与存储（storage）。执行内存主要用于shuffles、joins、sorts和aggregations，存储内存则用于缓存或者跨节点的内部数据传输。1.6之前，对于一个executor，内存都由以下部分构成：
1）ExecutionMemory
这片内存区域是为了解决shuffles、joins、sorts and aggregations过程中为了避免频繁IO需要的buffer。通过spark.shuffle.memoryFraction（默认0.2）配置
2）StorageMemory
这片内存区域是为了解决block cache（就是你显示调用rdd.cache，rdd.persist等方法），还有就是broadcast，以及task result的存储。可以通过参数spark.storage.memoryFraction（默认0.6）设置
3）otherMemory
给系统预留的，因为程序本身运行也是需要内存的（默认为0.2）

传统内存管理不足：
1）shuffle占用内存0.2*0.8，内存分配这么少，可能会将数据spill大磁盘，频繁的磁盘IO是很大的负担，storage内存占用0.6，主要是为了迭代处理。传统的spark内存分配对操作人的要求非常高。（shuffle分配内存：ShuffleMemoryManager,TaskMemoryManager,ExecutionMemoryManager）一个task获得全部的Execution的Memory，其他Task过来就没有内存了，只能等待
2）默认情况下，Task在线程中可能会占满整个内存，分片数据特别大的情况下就会出现这种情况，其他Task没有内存了，剩下的cores就空闲了，这是巨大的浪费。这也是人为操作的不当造成的； 3）MEMORY_AND_DISK_SER的storage方式，获得RDD的数据是一条条获取，iterator的方式。如果内存不够（spark.storage.unrollFraction），unroll的读取数据过程，就是看内存是否足够，如果足够，就下一条。unroll的space是从Storage的内存空间中获得的。unroll的方式失败，就会直接放磁盘；
4）默认情况下，Task在spill到磁盘之前，会将部分数据存放到内存上，如果获取不到内存，就不会执行。永无止境的等待，消耗CPU和内存；

在此基础上，Spark提出了UnifiedMemoryManager，不再分ExecutionMemory和Storage Memory，实际上还是分的，只不过是Execution Memory访问Storage Memory，Storage Memory也可以访问Execution Memory，如果内存不够，就会去借。

35.RDD中的reduceByKey与groupByKey哪个性能高

（1）当采用reduceByKeyt时，Spark可以在每个分区移动数据之前将待输出数据与一个共用的key结合。借助下图可以理解在reduceByKey里究竟发生了什么。 注意在数据对被搬移前同一机器上同样的key是怎样被组合的(reduceByKey中的lamdba函数)。然后lamdba函数在每个区上被再次调用来将所有值reduce成一个最终结果。整个过程如下：

（2）当采用groupByKey时，由于它不接收函数，spark只能先将所有的键值对(key-value pair)都移动，这样的后果是集群节点之间的开销很大，导致传输延时。整个过程如下：

因此，在对大数据进行复杂计算时，reduceByKey优于groupByKey。

36.简述HBase的读写过程

https://www.cnblogs.com/dummyly/p/10099395.html

写操作流程
(1) Client通过Zookeeper的调度，向RegionServer发出写数据请求，在Region中写数据。
(2) 数据被写入Region的MemStore，直到MemStore达到预设阈值。
(3) MemStore中的数据被Flush成一个StoreFile。
(4) 随着StoreFile文件的不断增多，当其数量增长到一定阈值后，触发Compact合并操作，将多个StoreFile合并成一个StoreFile，同时进行版本合并和数据删除。
(5) StoreFiles通过不断的Compact合并操作，逐步形成越来越大的StoreFile。
(6) 单个StoreFile大小超过一定阈值后，触发Split操作，把当前Region Split成2个新的Region。父Region会下线，新Split出的2个子Region会被HMaster分配到相应的RegionServer上，使得原先1个Region的压力得以分流到2个Region上。
可以看出HBase只有增添数据，所有的更新和删除操作都是在后续的Compact历程中举行的，使得用户的写操作只要进入内存就可以立刻返回，实现了HBase I/O的高机能。

读操作流程
(1) Client访问Zookeeper，查找-ROOT-表，获取.META.表信息。
(2) 从.META.表查找，获取存放目标数据的Region信息，从而找到对应的RegionServer。
(3) 通过RegionServer获取需要查找的数据。
(4) Regionserver的内存分为MemStore和BlockCache两部分，MemStore主要用于写数据，BlockCache主要用于读数据。读请求先到MemStore中查数据，查不到就到BlockCache中查，再查不到就会到StoreFile上读，并把读的结果放入BlockCache。
寻址过程：client–>Zookeeper–>-ROOT-表–>.META.表–>RegionServer–>Region–>client

Hbase寻址机制

1）客户端发一个get请求：
第一步，到zookeeper上查找root的路径，zookeeper返回root表的地址
第二步，客户端跟root所在的regionserver交互，查找.meta.表所在的regionserver，regionserver返回一个结果给客户端
第三步，跟mate所在的server交互，并查找数据的实际位置，meta所在的server返回一个数据的regionserver位置给客户端
第四步，跟数据所在regionserver交互，发送io请求，获取数据
第五步，regionserver首先在memstore中查找数据，如果有则直接返回，如果无则进行第六步
第六步，regionserver调用hdfs的api访问数据，根据bloomfilter得到数据的位置，查找实际的结果
第七步，将hdfs返回的结果写入memstore，并且返回数据给客户端

37.在2.5亿个整数中，找出不重复的整数，注意：内存不足以容纳2.5亿个整数

方案1：
采用2-Bitmap(每个数分配2bit，00表示不存在，01表示出现一次，10表示多次，11无意义)进行，共需内存2^32 * 2 bit=1 GB内存，还可以接受。然后扫描这2.5亿个整数，查看Bitmap中相对应位，如果是00变01，01变10，10保持不变。扫描完事后，查看bitmap，把对应位是01的整数输出即可。
方案2：
进行划分小文件的方法。然后在小文件中找出不重复的整数，并排序。然后再进行归并，注意去除重复的元素。

方案1详解：
将bit-map扩展一下，用2bit表示一个数即可，0表示未出现，1表示出现一次，2表示出现2次及以上，在遍历这些数的时候，如果对应位置的值是0，则将其置为1；如果是1，将其置为2；如果是2，则保持不变。或者我们不用2bit来进行表示，我们用两个bit-map即可模拟实现这个2bit-map，都是一样的道理。
int getval(const unsigned char& c,int num){
int i=0,j=0;
if(((0x1<<(2num))&c)==(0x1<<(2num)))
i=1;
if(((0x1<<(2num+1))&c)==(0x1<<(2num+1)))
j=1;
return 2*j+i;
}
void setval(unsigned char& c,int num,int val){

if(val1){
c= c|(0x1<<(2*num));
}else if(val2){
c= c&(~((0x1<<(2num))));
c= c|(0x1<<(2num+1));
}
}
void setbit(unsigned char* a,int num){
unsigned char* p=a;
for(int i=0;i<num/4;++i){
p++;
}
if(getval(p,num)==0){
setval(p,num%4,1);
}else if(getval(p,num)==1){
setval(p,num%4,2);
}
}
int main(int argc, char argv) {
unsigned char a[102410241024];
memset(a,0,sizeof(a));
FILE* file=fopen(“in.txt”,“r”);
unsigned uu=250000000;
char rn=’\n’;
for(unsigned int i=0;i<uu;++i){
int r;
fscanf(file,"%d",&r);
setbit(a,r);
// fwrite(&rn,1,1,file);
}
unsigned count=0;
unsigned char* p=a;
for(size_t i=0;i<102410241024;++i){
for(int j=0;j<4;++j){
if(getval(*p,j)==1)
count++;
}
p++;
}
cout<<count<<endl;
fclose(file);
return 0;
}

38.CDH和HDP的区别

CDH支持的存储组件更丰富
HDP支持的数据分析组件更丰富
HDP对多维分析及可视化有了支持，引入Druid和Superset
HDP的HBase数据使用Phoenix的jdbc查询；CDH的HBase数据使用映射Hive到Impala的jdbc查询，但分析数据可以存储Impala内部表，提高查询响应
多维分析Druid纳入集群，会方便管理；但可视化工具Superset可以单独安装使用
CDH没有时序数据库，HDP将Druid作为时序数据库使用

39.Java原子操作

https://blog.csdn.net/qq_30118563/article/details/90106741

40.Java封装、继承和多态

https://www.cnblogs.com/ygj0930/p/10830957.html

41.jvm模型

https://blog.csdn.net/qzqanzc/article/details/81008598
https://www.cnblogs.com/dingyingsi/p/3760447.html

42.Flume taildirSorce重复读取数据解决方法

https://blog.csdn.net/liguangzhu620/article/details/78907582

43.Flume如何保证数据不丢

https://blog.csdn.net/qq_44472134/article/details/104485110

44.Java 类加载过程

https://baijiahao.baidu.com/s?id=1636309817155065432&wfr=spider&for=pc
类从被加载到虚拟机内存中开始，到卸载出内存为止，它的整个生命周期包括：加载、验证、准备、解析、初始化、使用和卸载七个阶段。它们的顺序如下图所示：

https://blog.csdn.net/xinlingchengbao/article/details/88376479

45.spark task运行原理

https://blog.csdn.net/sinat_25306771/article/details/51451908

46.手写一个线程安全的单例

https://blog.csdn.net/liuchaoxuan/article/details/79840013
Double Check Locking 双检查锁机制（推荐）
为了达到线程安全，又能提高代码执行效率，我们这里可以采用DCL的双检查锁机制来完成，代码实现如下：
public class MySingleton {

//使用volatile关键字保其可见性  
volatile private static MySingleton instance = null;  
  
private MySingleton(){}  
   
public static MySingleton getInstance() {  
    try {    
        if(instance != null){//懒汉式   
              
        }else{  
            //创建实例之前可能会有一些准备性的耗时工作   
            Thread.sleep(300);  
            synchronized (MySingleton.class) {  
                if(instance == null){//二次检查  
                    instance = new MySingleton();  
                }  
            }  
        }   
    } catch (InterruptedException e) {   
        e.printStackTrace();  
    }  
    return instance;  
}  

}

47.设计模式

https://www.runoob.com/design-pattern/design-pattern-tutorial.html

48.impala 和 kudu 的适用场景，读写性能如何

https://blog.csdn.net/high2011/article/details/80988977

49.Kafka ack原理

Kafka的ack机制，指的是producer的消息发送确认机制，这直接影响到Kafka集群的吞吐量和消息可靠性。而吞吐量和可靠性就像硬币的两面，两者不可兼得，只能平衡。
ack有3个可选值，分别是1，0，-1。

ack=1，简单来说就是，producer只要收到一个分区副本成功写入的通知就认为推送消息成功了。这里有一个地方需要注意，这个副本必须是leader副本。只有leader副本成功写入了，producer才会认为消息发送成功。
注意，ack的默认值就是1。这个默认值其实就是吞吐量与可靠性的一个折中方案。生产上我们可以根据实际情况进行调整，比如如果你要追求高吞吐量，那么就要放弃可靠性。
ack=0，简单来说就是，producer发送一次就不再发送了，不管是否发送成功。
ack=-1，简单来说就是，producer只有收到分区内所有副本的成功写入的通知才认为推送消息成功了。

接下来我们分析一下ack=1的情况下，为什么消息也会丢失？
ack=1的情况下，producer只要收到分区leader成功写入的通知就会认为消息发送成功了。如果leader成功写入后，还没来得及把数据同步到follower节点就挂了，这时候消息就丢失了。

50.Flink TaskManager和Job Manager通信

Flink 整个系统主要由两个组件组成，分别为 JobManager 和 TaskManager，Flink 架构也遵循 Master - Slave 架构设计原则，JobManager 为 Master 节点，TaskManager 为 Worker （Slave）节点。
所有组件之间的通信都是借助于 Akka Framework，包括任务的状态以及 Checkpoint 触发等信息。

Client 客户端
客户端负责将任务提交到集群，与 JobManager 构建 Akka 连接，然后将任务提交到 JobManager，通过和 JobManager 之间进行交互获取任务执行状态。
客户端提交任务可以采用 CLI 方式或者通过使用 Flink WebUI 提交，也可以在应用程序中指定 JobManager 的 RPC 网络端口构建 ExecutionEnvironment 提交 Flink 应用。

JobManager
JobManager 负责整个 Flink 集群任务的调度以及资源的管理，从客户端中获取提交的应用，然后根据集群中 TaskManager 上 TaskSlot 的使用情况，为提交的应用分配相应的 TaskSlot 资源并命令 TaskManager 启动从客户端中获取的应用。
JobManager 相当于整个集群的 Master 节点，且整个集群有且只有一个活跃的 JobManager ，负责整个集群的任务管理和资源管理。
JobManager 和 TaskManager 之间通过 Actor System 进行通信，获取任务执行的情况并通过 Actor System 将应用的任务执行情况发送给客户端。
同时在任务执行的过程中，Flink JobManager 会触发 Checkpoint 操作，每个 TaskManager 节点 收到 Checkpoint 触发指令后，完成 Checkpoint 操作，所有的 Checkpoint 协调过程都是在 Fink JobManager 中完成。
当任务完成后，Flink 会将任务执行的信息反馈给客户端，并且释放掉 TaskManager 中的资源以供下一次提交任务使用。

TaskManager
TaskManager 相当于整个集群的 Slave 节点，负责具体的任务执行和对应任务在每个节点上的资源申请和管理。
客户端通过将编写好的 Flink 应用编译打包，提交到 JobManager，然后 JobManager 会根据已注册在 JobManager 中 TaskManager 的资源情况，将任务分配给有资源的 TaskManager节点，然后启动并运行任务。
TaskManager 从 JobManager 接收需要部署的任务，然后使用 Slot 资源启动 Task，建立数据接入的网络连接，接收数据并开始数据处理。同时 TaskManager 之间的数据交互都是通过数据流的方式进行的。
可以看出，Flink 的任务运行其实是采用多线程的方式，这和 MapReduce 多 JVM 进行的方式有很大的区别，Flink 能够极大提高 CPU 使用效率，在多个任务和 Task 之间通过 TaskSlot 方式共享系统资源，每个 TaskManager 中通过管理多个 TaskSlot 资源池进行对资源进行有效管理。

51.Flink 双流 join方式

https://blog.csdn.net/xianzhen376/article/details/89810958
Flink 系列博客
https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/stream/operators/joining.html

双流Join是Flink面试的高频问题。一般情况下说明以下几点就可以hold了：
Join大体分类只有两种：Window Join和Interval Join。Window Join又可以根据Window的类型细分出3种：Tumbling Window Join、Sliding Window Join、Session Widnow Join。
Windows类型的join都是利用window的机制，先将数据缓存在Window State中，当窗口触发计算时，执行join操作；
interval join也是利用state存储数据再处理，区别在于state中的数据有失效机制，依靠数据触发数据清理；
目前Stream join的结果是数据的笛卡尔积；
日常使用中的一些问题，数据延迟、window序列化相关。

从源码可以获得3个关键点：
Join后的流是CoGroupedStream；
Join后的数据处理类是CoGroupWindowFunction；
GoGroupWindowFunction中WrappedFunction是JoinCoGroupFunction；

52.Flink state 管理和 checkpoint 的流程

状态管理及容错机制
状态管理及容错机制介绍

状态管理的基本概念
1.什么是状态

首先举一个无状态计算的例子：消费延迟计算。假设现在有一个消息队列，消息队列中有一个生产者持续往消费队列写入消息，多个消费者分别从消息队列中读取消息。从图上可以看出，生产者已经写入 16 条消息，Offset 停留在 15 ；有 3 个消费者，有的消费快，而有的消费慢。消费快的已经消费了 13 条数据，消费者慢的才消费了 7、8 条数据。
如何实时统计每个消费者落后多少条数据，如图给出了输入输出的示例。可以了解到输入的时间点有一个时间戳，生产者将消息写到了某个时间点的位置，每个消费者同一时间点分别读到了什么位置。刚才也提到了生产者写入了 15 条，消费者分别读取了 10、7、12 条。那么问题来了，怎么将生产者、消费者的进度转换为右侧示意图信息呢？
consumer 0 落后了 5 条，consumer 1 落后了 8 条，consumer 2 落后了 3 条，根据 Flink 的原理，此处需进行 Map 操作。Map 首先把消息读取进来，然后分别相减，即可知道每个 consumer 分别落后了几条。Map 一直往下发，则会得出最终结果。

大家会发现，在这种模式的计算中，无论这条输入进来多少次，输出的结果都是一样的，因为单条输入中已经包含了所需的所有信息。消费落后等于生产者减去消费者。生产者的消费在单条数据中可以得到，消费者的数据也可以在单条数据中得到，所以相同输入可以得到相同输出，这就是一个无状态的计算。

相应的什么是有状态的计算？
以访问日志统计量的例子进行说明，比如当前拿到一个 Nginx 访问日志，一条日志表示一个请求，记录该请求从哪里来，访问的哪个地址，需要实时统计每个地址总共被访问了多少次，也即每个 API 被调用了多少次。可以看到下面简化的输入和输出，输入第一条是在某个时间点请求 GET 了 /api/a；第二条日志记录了某个时间点 Post /api/b ;第三条是在某个时间点 GET了一个 /api/a，总共有 3 个 Nginx 日志。从这 3 条 Nginx 日志可以看出，第一条进来输出 /api/a 被访问了一次，第二条进来输出 /api/b 被访问了一次，紧接着又进来一条访问 api/a，所以 api/a 被访问了 2 次。不同的是，两条 /api/a 的 Nginx 日志进来的数据是一样的，但输出的时候结果可能不同，第一次输出 count=1 ，第二次输出 count=2，说明相同输入可能得到不同输出。输出的结果取决于当前请求的 API 地址之前累计被访问过多少次。第一条过来累计是 0 次，count = 1，第二条过来 API 的访问已经有一次了，所以 /api/a 访问累计次数 count=2。单条数据其实仅包含当前这次访问的信息，而不包含所有的信息。要得到这个结果，还需要依赖 API 累计访问的量，即状态。

这个计算模式是将数据输入算子中，用来进行各种复杂的计算并输出数据。这个过程中算子会去访问之前存储在里面的状态。另外一方面，它还会把现在的数据对状态的影响实时更新，如果输入 200 条数据，最后输出就是 200 条结果。

什么场景会用到状态呢？下面列举了常见的 4 种：
去重：比如上游的系统数据可能会有重复，落到下游系统时希望把重复的数据都去掉。去重需要先了解哪些数据来过，哪些数据还没有来，也就是把所有的主键都记录下来，当一条数据到来后，能够看到在主键当中是否存在。
窗口计算：比如统计每分钟 Nginx 日志 API 被访问了多少次。窗口是一分钟计算一次，在窗口触发前，如 08:00 ~ 08:01 这个窗口，前59秒的数据来了需要先放入内存，即需要把这个窗口之内的数据先保留下来，等到 8:01 时一分钟后，再将整个窗口内触发的数据输出。未触发的窗口数据也是一种状态。
机器学习/深度学习：如训练的模型以及当前模型的参数也是一种状态，机器学习可能每次都用有一个数据集，需要在数据集上进行学习，对模型进行一个反馈。
访问历史数据：比如与昨天的数据进行对比，需要访问一些历史数据。如果每次从外部去读，对资源的消耗可能比较大，所以也希望把这些历史数据也放入状态中做对比。

2.为什么要管理状态

管理状态最直接的方式就是将数据都放到内存中，这也是很常见的做法。比如在做 WordCount 时，Word 作为输入，Count 作为输出。在计算的过程中把输入不断累加到 Count。

但对于流式作业有以下要求：
7*24小时运行，高可靠；
数据不丢不重，恰好计算一次；
数据实时产出，不延迟；

基于以上要求，内存的管理就会出现一些问题。由于内存的容量是有限制的。如果要做 24 小时的窗口计算，将 24 小时的数据都放到内存，可能会出现内存不足；另外，作业是 7*24，需要保障高可用，机器若出现故障或者宕机，需要考虑如何备份及从备份中去恢复，保证运行的作业不受影响；此外，考虑横向扩展，假如网站的访问量不高，统计每个 API 访问次数的程序可以用单线程去运行，但如果网站访问量突然增加，单节点无法处理全部访问数据，此时需要增加几个节点进行横向扩展，这时数据的状态如何平均分配到新增加的节点也问题之一。因此，将数据都放到内存中，并不是最合适的一种状态管理方式。

3.理想的状态管理

最理想的状态管理需要满足易用、高效、可靠三点需求：
易用，Flink 提供了丰富的数据结构、多样的状态组织形式以及简洁的扩展接口，让状态管理更加易用；
高效，实时作业一般需要更低的延迟，一旦出现故障，恢复速度也需要更快；当处理能力不够时，可以横向扩展，同时在处理备份时，不影响作业本身处理性能；
可靠，Flink 提供了状态持久化，包括不丢不重的语义以及具备自动的容错能力，比如 HA，当节点挂掉后会自动拉起，不需要人工介入。

Flink 状态的类型与使用示例

1. Managed State & Raw State

Managed State 是 Flink 自动管理的 State，而 Raw State 是原生态 State，两者的区别如下：
从状态管理方式的方式来说，Managed State 由 Flink Runtime 管理，自动存储，自动恢复，在内存管理上有优化；而 Raw State 需要用户自己管理，需要自己序列化，Flink 不知道 State 中存入的数据是什么结构，只有用户自己知道，需要最终序列化为可存储的数据结构。
从状态数据结构来说，Managed State 支持已知的数据结构，如 Value、List、Map 等。而 Raw State只支持字节数组 ，所有状态都要转换为二进制字节数组才可以。
从推荐使用场景来说，Managed State 大多数情况下均可使用，而 Raw State 是当 Managed State 不够用时，比如需要自定义 Operator 时，推荐使用 Raw State。

2. Keyed State & Operator State

Managed State 分为两种，一种是 Keyed State；另外一种是 Operator State。在Flink Stream模型中，Datastream 经过 keyBy 的操作可以变为 KeyedStream 。

每个 Key 对应一个 State，即一个 Operator 实例处理多个 Key，访问相应的多个 State，并由此就衍生了 Keyed State。Keyed State 只能用在 KeyedStream 的算子中，即在整个程序中没有 keyBy 的过程就没有办法使用 KeyedStream。

相比较而言，Operator State 可以用于所有算子，相对于数据源有一个更好的匹配方式，常用于 Source，例如 FlinkKafkaConsumer。相比 Keyed State，一个 Operator 实例对应一个 State，随着并发的改变，Keyed State 中，State 随着 Key 在实例间迁移，比如原来有 1 个并发，对应的 API 请求过来，/api/a 和 /api/b 都存放在这个实例当中；如果请求量变大，需要扩容，就会把 /api/a 的状态和 /api/b 的状态分别放在不同的节点。由于 Operator State 没有 Key，并发改变时需要选择状态如何重新分配。其中内置了 2 种分配方式：一种是均匀分配，另外一种是将所有 State 合并为全量 State 再分发给每个实例。

在访问上，Keyed State 通过 RuntimeContext 访问，这需要 Operator 是一个Rich Function。Operator State 需要自己实现 CheckpointedFunction 或 ListCheckpointed 接口。在数据结构上，Keyed State 支持的数据结构，比如 ValueState、ListState、ReducingState、AggregatingState 和 MapState；而 Operator State 支持的数据结构相对较少，如 ListState。

3. Keyed State 使用示例

Keyed State 有很多种，如图为几种 Keyed State 之间的关系。首先 State 的子类中一级子类有 ValueState、MapState、AppendingState。AppendingState 又有一个子类 MergingState。MergingState 又分为 3 个子类分别是ListState、ReducingState、AggregatingState。这个继承关系使它们的访问方式、数据结构也存在差异。

几种 Keyed State 的差异具体体现在：
ValueState 存储单个值，比如 Wordcount，用 Word 当 Key，State 就是它的 Count。这里面的单个值可能是数值或者字符串，作为单个值，访问接口可能有两种，get 和 set。在 State 上体现的是 update(T) / T value()。
MapState 的状态数据类型是 Map，在 State 上有 put、remove等。需要注意的是在 MapState 中的 key 和 Keyed state 中的 key 不是同一个。
ListState 状态数据类型是 List，访问接口如 add、update 等。
ReducingState 和 AggregatingState 与 ListState 都是同一个父类，但状态数据类型上是单个值，原因在于其中的 add 方法不是把当前的元素追加到列表中，而是把当前元素直接更新进了 Reducing 的结果中。
AggregatingState 的区别是在访问接口，ReducingState 中 add（T）和 T get() 进去和出来的元素都是同一个类型，但在 AggregatingState 输入的 IN，输出的是 OUT。

下面以 ValueState 为例，来阐述一下具体如何使用，以状态机的案例来讲解 。

源代码地址：https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/statemachine/StateMachineExample.java

感兴趣的同学可直接查看完整源代码，在此截取部分。如图为 Flink 作业的主方法与主函数中的内容，前面的输入、后面的输出以及一些个性化的配置项都已去掉，仅保留了主干。

首先 events 是一个 DataStream，通过 env.addSource 加载数据进来，接下来有一个 DataStream 叫 alerts，先 keyby 一个 sourceAddress，然后在 flatMap 一个StateMachineMapper。StateMachineMapper 就是一个状态机，状态机指有不同的状态与状态间有不同的转换关系的结合，以买东西的过程简单举例。首先下订单，订单生成后状态为待付款，当再来一个事件状态付款成功，则事件的状态将会从待付款变为已付款，待发货。已付款，待发货的状态再来一个事件发货，订单状态将会变为配送中，配送中的状态再来一个事件签收，则该订单的状态就变为已签收。在整个过程中，随时都可以来一个事件，取消订单，无论哪个状态，一旦触发了取消订单事件最终就会将状态转移到已取消，至此状态就结束了。

Flink 写状态机是如何实现的？首先这是一个 RichFlatMapFunction，要用 Keyed State getRuntimeContext，getRuntimeContext 的过程中需要 RichFunction，所以需要在 open 方法中获取 currentState ，然后 getState，currentState 保存的是当前状态机上的状态。

如果刚下订单，那么 currentState 就是待付款状态，初始化后，currentState 就代表订单完成。订单来了后，就会走 flatMap 这个方法，在 flatMap 方法中，首先定义一个 State，从 currentState 取出，即 Value，Value 取值后先判断值是否为空，如果 sourceAddress state 是空，则说明没有被使用过，那么此状态应该为刚创建订单的初始状态，即待付款。然后赋值 state = State.Initial，注意此处的 State 是本地的变量，而不是 Flink 中管理的状态，将它的值从状态中取出。接下来在本地又会来一个变量，然后 transition，将事件对它的影响加上，刚才待付款的订单收到付款成功的事件，就会变成已付款，待发货，然后 nextState 即可算出。此外，还需要判断 State 是否合法，比如一个已签收的订单，又来一个状态叫取消订单，会发现已签收订单不能被取消，此时这个状态就会下发，订单状态为非法状态。

如果不是非法的状态，还要看该状态是否已经无法转换，比如这个状态变为已取消时，就不会在有其他的状态再发生了，此时就会从 state 中 clear。clear 是所有的 Flink 管理 keyed state 都有的公共方法，意味着将信息删除，如果既不是一个非法状态也不是一个结束状态，后面可能还会有更多的转换，此时需要将订单的当前状态 update ，这样就完成了 ValueState 的初始化、取值、更新以及清零，在整个过程中状态机的作用就是将非法的状态进行下发，方便下游进行处理。其他的状态也是类似的使用方式。

容错机制与故障恢复

1. 状态如何保存及恢复
   Flink 状态保存主要依靠 Checkpoint 机制，Checkpoint 会定时制作分布式快照，对程序中的状态进行备份。分布式快照是如何实现的可以参考【第二课时】的内容，这里就不再阐述分布式快照具体是如何实现的。分布式快照 Checkpoint 完成后，当作业发生故障了如何去恢复？假如作业分布跑在 3 台机器上，其中一台挂了。这个时候需要把进程或者线程移到 active 的 2 台机器上，此时还需要将整个作业的所有 Task 都回滚到最后一次成功 Checkpoint 中的状态，然后从该点开始继续处理。

如果要从 Checkpoint 恢复，必要条件是数据源需要支持数据重新发送。Checkpoint恢复后， Flink 提供两种一致性语义，一种是恰好一次，一种是至少一次。在做 Checkpoint时，可根据 Barries 对齐来判断是恰好一次还是至少一次，如果对齐，则为恰好一次，否则没有对齐即为至少一次。如果只有一个上游，也就是说 Barries 是不需要对齐的的；如果只有一个 Checkpoint 在做，不管什么时候从 Checkpoint 恢复，都会恢复到刚才的状态；如果有多个上游，假如一个上游的 Barries 到了，另一个 Barries 还没有来，如果这个时候对状态进行快照，那么从这个快照恢复的时候其中一个上游的数据可能会有重复。

Checkpoint 通过代码的实现方法如下：
首先从作业的运行环境 env.enableCheckpointing 传入 1000，意思是做 2 个 Checkpoint 的事件间隔为 1 秒。Checkpoint 做的越频繁，恢复时追数据就会相对减少，同时 Checkpoint 相应的也会有一些 IO 消耗。
接下来是设置 Checkpoint 的 model，即设置了 Exactly_Once 语义，表示需要 Barrier 对齐，这样可以保证消息不会丢失也不会重复。
setMinPauseBetweenCheckpoints 是 2 个 Checkpoint 之间最少是要等 500ms，也就是刚做完一个 Checkpoint。比如某个 Checkpoint 做了700ms，按照原则过 300ms 应该是做下一个 Checkpoint，因为设置了 1000ms 做一次 Checkpoint 的，但是中间的等待时间比较短，不足 500ms 了，需要多等 200ms，因此以这样的方式防止 Checkpoint 太过于频繁而导致业务处理的速度下降。
setCheckpointTimeout 表示做 Checkpoint 多久超时，如果 Checkpoint 在 1min 之内尚未完成，说明 Checkpoint 超时失败。
setMaxConcurrentCheckpoints 表示同时有多少个 Checkpoint 在做快照，这个可以根据具体需求去做设置。
enableExternalizedCheckpoints 表示下 Cancel 时是否需要保留当前的 Checkpoint，默认 Checkpoint 会在整个作业 Cancel 时被删除。Checkpoint 是作业级别的保存点。

上面讲过，除了故障恢复之外，还需要可以手动去调整并发重新分配这些状态。手动调整并发，必须要重启作业并会提示 Checkpoint 已经不存在，那么作业如何恢复数据？

一方面 Flink 在 Cancel 时允许在外部介质保留 Checkpoint ；另一方面，Flink 还有另外一个机制是 SavePoint。

Savepoint 与 Checkpoint 类似，同样是把状态存储到外部介质。当作业失败时，可以从外部恢复。Savepoint 与 Checkpoint 有什么区别呢？
从触发管理方式来讲，Checkpoint 由 Flink 自动触发并管理，而 Savepoint 由用户手动触发并人肉管理；
从用途来讲，Checkpoint 在 Task 发生异常时快速恢复，例如网络抖动或超时异常，而 Savepoint 有计划地进行备份，使作业能停止后再恢复，例如修改代码、调整并发；
最后从特点来讲，Checkpoint 比较轻量级，作业出现问题会自动从故障中恢复，在作业停止后默认清除；而 Savepoint 比较持久，以标准格式存储，允许代码或配置发生改变，恢复需要启动作业手动指定一个路径恢复。

2. 可选的状态存储方式

Checkpoint 的存储，第一种是内存存储，即 MemoryStateBackend，构造方法是设置最大的StateSize，选择是否做异步快照，这种存储状态本身存储在 TaskManager 节点也就是执行节点内存中的，因为内存有容量限制，所以单个 State maxStateSize 默认 5 M，且需要注意 maxStateSize <= akka.framesize 默认 10 M。Checkpoint 存储在 JobManager 内存中，因此总大小不超过 JobManager 的内存。推荐使用的场景为：本地测试、几乎无状态的作业，比如 ETL、JobManager 不容易挂，或挂掉影响不大的情况。不推荐在生产场景使用。

另一种就是在文件系统上的 FsStateBackend ，构建方法是需要传一个文件路径和是否异步快照。State 依然在 TaskManager 内存中，但不会像 MemoryStateBackend 有 5 M 的设置上限，Checkpoint 存储在外部文件系统（本地或 HDFS），打破了总大小 Jobmanager 内存的限制。容量限制上，单 TaskManager 上 State 总量不超过它的内存，总大小不超过配置的文件系统容量。推荐使用的场景、常规使用状态的作业、例如分钟级窗口聚合或 join、需要开启HA的作业。

还有一种存储为 RocksDBStateBackend ，RocksDB 是一个 key/value 的内存存储系统，和其他的 key/value 一样，先将状态放到内存中，如果内存快满时，则写入到磁盘中，但需要注意 RocksDB 不支持同步的 Checkpoint，构造方法中没有同步快照这个选项。不过 RocksDB 支持增量的 Checkpoint，也是目前唯一增量 Checkpoint 的 Backend，意味着并不需要把所有 sst 文件上传到 Checkpoint 目录，仅需要上传新生成的 sst 文件即可。它的 Checkpoint 存储在外部文件系统（本地或HDFS），其容量限制只要单个 TaskManager 上 State 总量不超过它的内存+磁盘，单 Key最大 2G，总大小不超过配置的文件系统容量即可。推荐使用的场景为：超大状态的作业，例如天级窗口聚合、需要开启 HA 的作业、最好是对状态读写性能要求不高的作业。

总结

1. 为什么要使用状态？
   前面提到有状态的作业要有有状态的逻辑，有状态的逻辑是因为数据之间存在关联，单条数据是没有办法把所有的信息给表现出来。所以需要通过状态来满足业务逻辑。
   2.为什么要管理状态？
   使用了状态，为什么要管理状态？因为实时作业需要7*24不间断的运行，需要应对不可靠的因素而带来的影响。
   3.如何选择状态的类型和存储方式？
   那如何选择状态的类型和存储方式？结合前面的内容，可以看到，首先是要分析清楚业务场景；比如想要做什么，状态到底大不大。比较各个方案的利弊，选择根据需求合适的状态类型和存储方式即可。

53.Flink分层架构

Flink 结构与支持
Flink是一个分层结构的系统，不同层的栈建立在其下层基础上，并且每一层所包含的组件都提供了特定的抽象，用来服务于上层组件。Flink分层的组件栈如下图所示：

1.部署：Flink支持本地运行、能在独立集群或者在被Yarn或Mesos管理的集群上运行，也能部署在云上
2.运行：Flink的核心是分布式流式数据引擎，意味着数据以一次一个事件的形式被处理
3.API：DataSteam、DataSet、Table、SQL API
4.扩展库：Flink还包括用于复杂事件处理，机器学习，图形处理和Apache Storm兼容性的专用代码库

Deployment调度层
该层主要涉及了Flink的部署模式，Flink支持多种部署模式：本地、集群（Standalone/Yarn）、云（GCE/EC2）。Standalone部署模式与Spark类似，Flink on Yarn的部署模式如下图所示：

了解Yarn的话，对上图的原理非常熟悉，实际Flink也实现了，满足在Yarn集群上运行的各个组件：Flink Yarn Client负责与Yarn RM通信协商资源请求，Flink JobManager和Flink TaskManager分别申请到Container去运行各自的进程。通过上图可以看到，Yarn AM与Flink JobManager在同一个Container中，这样AM可以知道Flink JobManager的地址，从而AM可以申请Container去启动Flink TaskManager。待Flink成功运行在Yarn集群上，Flink Yarn Client就可以提交Flink Job到Flink JobManager，并进行后续的映射、调度和计算处理。

Runtime
Runtime层提供了支持Flink计算的全部核心实现，比如支持分布式Stream处理、JobGraph到ExecutionGraph的映射、调度等等，为上层API层提供基础服务。
分布式流处理引擎

API
API层主要实现了面向*Stream的流处理和面向Batch的批处理API，其中面向流处理对应DataStream API，面向批处理对应DataSet API。

Libraries
Libraries层也可以称为Flink应用架构层，根据API层的划分，在API层之上构建的满足特定应用的计算架构，也分别对应于面向流处理和面向批处理两类。
面向流处理支持：CEP（复杂事件处理）、基于SQL-like的操作（基于Table的关系操作）
面向批处理支持：FlinkML（机器学习库）、Gelly（图处理）
运行时层以JobGraph形式接收程序。JobGraph即为一个一般化的并行数据流图（data flow），它拥有任意数量的Task来接收和产生data stream
Datastream API 和 DataSet API都会使用单独编译的处理方式生成JobGraph。DataSet API使用Optimizer优化器来决定针对程序的优化方法，而datastream API 则使用stream builder流生成器来完成该任务。
在执行JobGraph时Flink提供了多种候选部署方案（如local、remote、yarn等）
Flink附随了一些产生DataSet或DataStream API程序的类库和API：处理学习的FlinkML，图像处理的Gelly，复杂事件处理的CEP。

54.Flink窗口

Flink Time & Window解析

55.阐述Flink如何处理反压，相比Storm、Spark Streaming提供的反压机制，描述其实现有什么不同？

网络流控及反压剖析
Flink 原理与实现：如何处理反压问题
如何分析和处理Flink反压？
sparkstreaming的反压机制

流处理系统需要能优雅地处理反压（backpressure）问题。反压通常产生于这样的场景：短时负载高峰导致系统接收数据的速率远高于它处理数据的速率。许多日常问题都会导致反压，例如，垃圾回收停顿可能会导致流入的数据快速堆积，或者遇到大促或秒杀活动导致流量陡增。反压如果不能得到正确的处理，可能会导致资源耗尽甚至系统崩溃。

目前主流的流处理系统 Storm/JStorm/Spark Streaming/Flink 都已经提供了反压机制，不过其实现各不相同。
Storm
Storm 是通过监控 Bolt 中的接收队列负载情况，如果超过高水位值就会将反压信息写到 Zookeeper ，Zookeeper 上的 watch 会通知该拓扑的所有 Worker 都进入反压状态，最后 Spout 停止发送 tuple。具体实现可以看这个 JIRA STORM-886。
JStorm
JStorm 认为直接停止 Spout 的发送太过暴力，存在大量问题。当下游出现阻塞时，上游停止发送，下游消除阻塞后，上游又开闸放水，过了一会儿，下游又阻塞，上游又限流，如此反复，整个数据流会一直处在一个颠簸状态。所以 JStorm 是通过逐级降速来进行反压的，效果会较 Storm 更为稳定，但算法也更复杂。另外 JStorm 没有引入 Zookeeper 而是通过 TopologyMaster 来协调拓扑进入反压状态，这降低了 Zookeeper 的负载。
Spark Streaming
Spark Streaming程序中当计算过程中出现batch processing time > batch interval的情况时，(其中batch processing time为实际计算一个批次花费时间，batch interval为Streaming应用设置的批处理间隔),意味着处理数据的速度小于接收数据的速度，如果这种情况持续过长的时间，会造成数据在内存中堆积，导致Receiver所在Executor内存溢出等问题(如果设置StorageLevel包含disk, 则内存存放不下的数据会溢写至disk, 加大延迟),可以通过设置参数spark.streaming.receiver.maxRate来限制Receiver的数据接收速率，此举虽然可以通过限制接收速率，来适配当前的处理能力，防止内存溢出，但也会引入其它问题。比如：producer数据生产高于maxRate，当前集群处理能力也高于maxRate，这就会造成资源利用率下降等问题。为了更好的协调数据接收速率与资源处理能力，Spark Streaming 从v1.5开始引入反压机制（back-pressure）,通过动态控制数据接收速率来适配集群数据处理能力。
Spark Streaming Backpressure: 根据JobScheduler反馈作业的执行信息来动态调整Receiver数据接收率。通过属性"spark.streaming.backpressure.enabled"来控制是否启用backpressure机制，默认值false，即不启用。

Flink中的反压
那么 Flink 是怎么处理反压的呢？
答案非常简单：Flink 没有使用任何复杂的机制来解决反压问题，因为根本不需要那样的方案！它利用自身作为纯数据流引擎的优势来优雅地响应反压问题。下面我们会深入分析 Flink 是如何在 Task 之间传输数据的，以及数据流如何实现自然降速的。

Flink 在运行时主要由 operators 和 streams 两大组件构成。每个 operator 会消费中间态的流，并在流上进行转换，然后生成新的流。对于 Flink 的网络机制一种形象的类比是，Flink 使用了高效有界的分布式阻塞队列，就像 Java 通用的阻塞队列（BlockingQueue）一样。还记得经典的线程间通信案例：生产者消费者模型吗？使用 BlockingQueue 的话，一个较慢的接受者会降低发送者的发送速率，因为一旦队列满了（有界队列）发送者会被阻塞。Flink 解决反压的方案就是这种感觉。

在 Flink 中，这些分布式阻塞队列就是这些逻辑流，而队列容量是通过缓冲池（LocalBufferPool）来实现的。每个被生产和被消费的流都会被分配一个缓冲池。缓冲池管理着一组缓冲(Buffer)，缓冲在被消费后可以被回收循环利用。这很好理解：你从池子中拿走一个缓冲，填上数据，在数据消费完之后，又把缓冲还给池子，之后你可以再次使用它。

在解释 Flink 的反压原理之前，我们必须先对 Flink 中网络传输的内存管理有个了解。

网络传输中的内存管理
如下图所示展示了 Flink 在网络传输场景下的内存管理。网络上传输的数据会写到 Task 的 InputGate（IG） 中，经过 Task 的处理后，再由 Task 写到 ResultPartition（RS） 中。每个 Task 都包括了输入和输入，输入和输出的数据存在 Buffer 中（都是字节数据）。Buffer 是 MemorySegment 的包装类。

TaskManager（TM）在启动时，会先初始化NetworkEnvironment对象，TM 中所有与网络相关的东西都由该类来管理（如 Netty 连接），其中就包括NetworkBufferPool。根据配置，Flink 会在 NetworkBufferPool 中生成一定数量（默认2048）的内存块 MemorySegment（关于 Flink 的内存管理，后续文章会详细谈到），内存块的总数量就代表了网络传输中所有可用的内存。NetworkEnvironment 和 NetworkBufferPool 是 Task 之间共享的，每个 TM 只会实例化一个。

Task 线程启动时，会向 NetworkEnvironment 注册，NetworkEnvironment 会为 Task 的 InputGate（IG）和 ResultPartition（RP） 分别创建一个 LocalBufferPool（缓冲池）并设置可申请的 MemorySegment（内存块）数量。IG 对应的缓冲池初始的内存块数量与 IG 中 InputChannel 数量一致，RP 对应的缓冲池初始的内存块数量与 RP 中的 ResultSubpartition 数量一致。不过，每当创建或销毁缓冲池时，NetworkBufferPool 会计算剩余空闲的内存块数量，并平均分配给已创建的缓冲池。注意，这个过程只是指定了缓冲池所能使用的内存块数量，并没有真正分配内存块，只有当需要时才分配。为什么要动态地为缓冲池扩容呢？因为内存越多，意味着系统可以更轻松地应对瞬时压力（如GC），不会频繁地进入反压状态，所以我们要利用起那部分闲置的内存块。

在 Task 线程执行过程中，当 Netty 接收端收到数据时，为了将 Netty 中的数据拷贝到 Task 中，InputChannel（实际是 RemoteInputChannel）会向其对应的缓冲池申请内存块（上图中的①）。如果缓冲池中也没有可用的内存块且已申请的数量还没到池子上限，则会向 NetworkBufferPool 申请内存块（上图中的②）并交给 InputChannel 填上数据（上图中的③和④）。如果缓冲池已申请的数量达到上限了呢？或者 NetworkBufferPool 也没有可用内存块了呢？这时候，Task 的 Netty Channel 会暂停读取，上游的发送端会立即响应停止发送，拓扑会进入反压状态。当 Task 线程写数据到 ResultPartition 时，也会向缓冲池请求内存块，如果没有可用内存块时，会阻塞在请求内存块的地方，达到暂停写入的目的。

当一个内存块被消费完成之后（在输入端是指内存块中的字节被反序列化成对象了，在输出端是指内存块中的字节写入到 Netty Channel 了），会调用 Buffer.recycle() 方法，会将内存块还给 LocalBufferPool （上图中的⑤）。如果LocalBufferPool中当前申请的数量超过了池子容量（由于上文提到的动态容量，由于新注册的 Task 导致该池子容量变小），则LocalBufferPool会将该内存块回收给 NetworkBufferPool（上图中的⑥）。如果没超过池子容量，则会继续留在池子中，减少反复申请的开销。

反压的过程
下面这张图简单展示了两个 Task 之间的数据传输以及 Flink 如何感知到反压的：

记录“A”进入了 Flink 并且被 Task 1 处理。（这里省略了 Netty 接收、反序列化等过程）
记录被序列化到 buffer 中。
该 buffer 被发送到 Task 2，然后 Task 2 从这个 buffer 中读出记录。

不要忘了：记录能被 Flink 处理的前提是，必须有空闲可用的 Buffer。

结合上面两张图看：Task 1 在输出端有一个相关联的 LocalBufferPool（称缓冲池1），Task 2 在输入端也有一个相关联的 LocalBufferPool（称缓冲池2）。如果缓冲池1中有空闲可用的 buffer 来序列化记录 “A”，我们就序列化并发送该 buffer。
这里我们需要注意两个场景：
本地传输：如果 Task 1 和 Task 2 运行在同一个 worker 节点（TaskManager），该 buffer 可以直接交给下一个 Task。一旦 Task 2 消费了该 buffer，则该 buffer 会被缓冲池1回收。如果 Task 2 的速度比 1 慢，那么 buffer 回收的速度就会赶不上 Task 1 取 buffer 的速度，导致缓冲池1无可用的 buffer，Task 1 等待在可用的 buffer 上。最终形成 Task 1 的降速。
远程传输：如果 Task 1 和 Task 2 运行在不同的 worker 节点上，那么 buffer 会在发送到网络（TCP Channel）后被回收。在接收端，会从 LocalBufferPool 中申请 buffer，然后拷贝网络中的数据到 buffer 中。如果没有可用的 buffer，会停止从 TCP 连接中读取数据。在输出端，通过 Netty 的水位值机制来保证不往网络中写入太多数据（后面会说）。如果网络中的数据（Netty输出缓冲中的字节数）超过了高水位值，我们会等到其降到低水位值以下才继续写入数据。这保证了网络中不会有太多的数据。如果接收端停止消费网络中的数据（由于接收端缓冲池没有可用 buffer），网络中的缓冲数据就会堆积，那么发送端也会暂停发送。另外，这会使得发送端的缓冲池得不到回收，writer 阻塞在向 LocalBufferPool 请求 buffer，阻塞了 writer 往 ResultSubPartition 写数据。

这种固定大小缓冲池就像阻塞队列一样，保证了 Flink 有一套健壮的反压机制，使得 Task 生产数据的速度不会快于消费的速度。我们上面描述的这个方案可以从两个 Task 之间的数据传输自然地扩展到更复杂的 pipeline 中，保证反压机制可以扩散到整个 pipeline。

Netty 水位值机制
下方的代码是初始化 NettyServer 时配置的水位值参数。
// 默认高水位值为2个buffer大小, 当接收端消费速度跟不上，发送端会立即感知到
bootstrap.childOption(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, config.getMemorySegmentSize() + 1);
bootstrap.childOption(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, 2 * config.getMemorySegmentSize());

当输出缓冲中的字节数超过了高水位值, 则 Channel.isWritable() 会返回false。当输出缓存中的字节数又掉到了低水位值以下, 则 Channel.isWritable() 会重新返回true。Flink 中发送数据的核心代码在 PartitionRequestQueue 中，该类是 server channel pipeline 的最后一层。发送数据关键代码如下所示。
private void writeAndFlushNextMessageIfPossible(final Channel channel) throws IOException {
if (fatalError) {
return;
}

Buffer buffer = null;

try {
// channel.isWritable() 配合 WRITE_BUFFER_LOW_WATER_MARK
// 和 WRITE_BUFFER_HIGH_WATER_MARK 实现发送端的流量控制
if (channel.isWritable()) {
// 注意: 一个while循环也就最多只发送一个BufferResponse, 连续发送BufferResponse是通过writeListener回调实现的
while (true) {
if (currentPartitionQueue == null && (currentPartitionQueue = queue.poll()) == null) {
return;
}

    buffer = currentPartitionQueue.getNextBuffer();

    if (buffer == null) {
      // 跳过这部分代码
      ...
    }
    else {
      // 构造一个response返回给客户端
      BufferResponse resp = new BufferResponse(buffer, currentPartitionQueue.getSequenceNumber(), currentPartitionQueue.getReceiverId());

      if (!buffer.isBuffer() &&
          EventSerializer.fromBuffer(buffer, getClass().getClassLoader()).getClass() == EndOfPartitionEvent.class) {
        // 跳过这部分代码。batch 模式中 subpartition 的数据准备就绪，通知下游消费者。
        ...
      }

      // 将该response发到netty channel, 当写成功后, 
      // 通过注册的writeListener又会回调进来, 从而不断地消费 queue 中的请求
      channel.writeAndFlush(resp).addListener(writeListener);

      return;
    }
  }
}

}
catch (Throwable t) {
if (buffer != null) {
buffer.recycle();
}

throw new IOException(t.getMessage(), t);

}
}

// 当水位值降下来后（channel 再次可写），会重新触发发送函数
@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
writeAndFlushNextMessageIfPossible(ctx.channel());
}

核心发送方法中如果channel不可写，则会跳过发送。当channel再次可写后，Netty 会调用该Handle的 channelWritabilityChanged 方法，从而重新触发发送函数。

反压实验
另外，官方博客中为了展示反压的效果，给出了一个简单的实验。下面这张图显示了：随着时间的改变，生产者（黄色线）和消费者（绿色线）每5秒的平均吞吐与最大吞吐（在单一JVM中每秒达到8百万条记录）的百分比。我们通过衡量task每5秒钟处理的记录数来衡量平均吞吐。该实验运行在单 JVM 中，不过使用了完整的 Flink 功能栈。

首先，我们运行生产task到它最大生产速度的60%（我们通过Thread.sleep()来模拟降速）。消费者以同样的速度处理数据。然后，我们将消费task的速度降至其最高速度的30%。你就会看到背压问题产生了，正如我们所见，生产者的速度也自然降至其最高速度的30%。接着，停止消费task的人为降速，之后生产者和消费者task都达到了其最大的吞吐。接下来，我们再次将消费者的速度降至30%，pipeline给出了立即响应：生产者的速度也被自动降至30%。最后，我们再次停止限速，两个task也再次恢复100%的速度。总而言之，我们可以看到：生产者和消费者在 pipeline 中的处理都在跟随彼此的吞吐而进行适当的调整，这就是我们希望看到的反压的效果。

反压监控
在 Storm/JStorm 中，只要监控到队列满了，就可以记录下拓扑进入反压了。但是 Flink 的反压太过于天然了，导致我们无法简单地通过监控队列来监控反压状态。Flink 在这里使用了一个 trick 来实现对反压的监控。如果一个 Task 因为反压而降速了，那么它会卡在向 LocalBufferPool 申请内存块上。那么这时候，该 Task 的 stack trace 就会长下面这样：
java.lang.Object.wait(Native Method)
o.a.f.[…].LocalBufferPool.requestBuffer(LocalBufferPool.java:163)
o.a.f.[…].LocalBufferPool.requestBufferBlocking(LocalBufferPool.java:133) <— BLOCKING request
[…]

那么事情就简单了。通过不断地采样每个 task 的 stack trace 就可以实现反压监控。

Flink 的实现中，只有当 Web 页面切换到某个 Job 的 Backpressure 页面，才会对这个 Job 触发反压检测，因为反压检测还是挺昂贵的。JobManager 会通过 Akka 给每个 TaskManager 发送TriggerStackTraceSample消息。默认情况下，TaskManager 会触发100次 stack trace 采样，每次间隔 50ms（也就是说一次反压检测至少要等待5秒钟）。并将这 100 次采样的结果返回给 JobManager，由 JobManager 来计算反压比率（反压出现的次数/采样的次数），最终展现在 UI 上。UI 刷新的默认周期是一分钟，目的是不对 TaskManager 造成太大的负担。

总结
Flink 不需要一种特殊的机制来处理反压，因为 Flink 中的数据传输相当于已经提供了应对反压的机制。因此，Flink 所能获得的最大吞吐量由其 pipeline 中最慢的组件决定。相对于 Storm/JStorm 的实现，Flink 的实现更为简洁优雅，源码中也看不见与反压相关的代码，无需 Zookeeper/TopologyMaster 的参与也降低了系统的负载，也利于对反压更迅速的响应。

56.Spark Streaming 反压（Back Pressure）机制介绍

背景
在默认情况下，Spark Streaming 通过 receivers (或者是 Direct 方式) 以生产者生产数据的速率接收数据。当 batch processing time > batch interval 的时候，也就是每个批次数据处理的时间要比 Spark Streaming 批处理间隔时间长；越来越多的数据被接收，但是数据的处理速度没有跟上，导致系统开始出现数据堆积，可能进一步导致 Executor 端出现 OOM 问题而出现失败的情况。
而在 Spark 1.5 版本之前，为了解决这个问题，对于 Receiver-based 数据接收器，我们可以通过配置 spark.streaming.receiver.maxRate 参数来限制每个 receiver 每秒最大可以接收的记录的数据；对于 Direct Approach 的数据接收，我们可以通过配置 spark.streaming.kafka.maxRatePerPartition 参数来限制每次作业中每个 Kafka 分区最多读取的记录条数。这种方法虽然可以通过限制接收速率，来适配当前的处理能力，但这种方式存在以下几个问题：
我们需要事先估计好集群的处理速度以及消息数据的产生速度；
这两种方式需要人工参与，修改完相关参数之后，我们需要手动重启 Spark Streaming 应用程序；
如果当前集群的处理能力高于我们配置的 maxRate，而且 producer 产生的数据高于 maxRate，这会导致集群资源利用率低下，而且也会导致数据不能够及时处理。

如果想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号：iteblog_hadoop
反压机制
那么有没有可能不需要人工干预，Spark Streaming 系统自动处理这些问题呢？当然有了！Spark 1.5 引入了反压（Back Pressure）机制，其通过动态收集系统的一些数据来自动地适配集群数据处理能力。详细的记录请参见 SPARK-7398 里面的说明。

Spark Streaming 1.5 以前的体系结构
在 Spark 1.5 版本之前，Spark Streaming 的体系结构如下所示：
Spark 数据堆积

如果想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号：iteblog_hadoop
数据是源源不断的通过 receiver 接收，当数据被接收后，其将这些数据存储在 Block Manager 中；为了不丢失数据，其还将数据备份到其他的 Block Manager 中；
Receiver Tracker 收到被存储的 Block IDs，然后其内部会维护一个时间到这些 block IDs 的关系；
Job Generator 会每隔 batchInterval 的时间收到一个事件，其会生成一个 JobSet；
Job Scheduler 运行上面生成的 JobSet。

Spark Streaming 1.5 之后的体系结构

如果想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号：iteblog_hadoop
为了实现自动调节数据的传输速率，在原有的架构上新增了一个名为 RateController 的组件，这个组件继承自 StreamingListener，其监听所有作业的 onBatchCompleted 事件，并且基于 processingDelay 、schedulingDelay 、当前 Batch 处理的记录条数以及处理完成事件来估算出一个速率；这个速率主要用于更新流每秒能够处理的最大记录的条数。速率估算器（RateEstimator）可以又多种实现，不过目前的 Spark 2.2 只实现了基于 PID 的速率估算器。
InputDStreams 内部的 RateController 里面会存下计算好的最大速率，这个速率会在处理完 onBatchCompleted 事件之后将计算好的速率推送到 ReceiverSupervisorImpl，这样接收器就知道下一步应该接收多少数据了。
如果用户配置了 spark.streaming.receiver.maxRate 或 spark.streaming.kafka.maxRatePerPartition，那么最后到底接收多少数据取决于三者的最小值。也就是说每个接收器或者每个 Kafka 分区每秒处理的数据不会超过 spark.streaming.receiver.maxRate 或 spark.streaming.kafka.maxRatePerPartition 的值。
详细的过程如下图所示：

如果想及时了解Spark、Hadoop或者Hbase相关的文章，欢迎关注微信公共帐号：iteblog_hadoop

Spark Streaming 反压机制的使用
在 Spark 启用反压机制很简单，只需要将 spark.streaming.backpressure.enabled 设置为 true 即可，这个参数的默认值为 false。反压机制还涉及以下几个参数，包括文档中没有列出来的：
spark.streaming.backpressure.initialRate： 启用反压机制时每个接收器接收第一批数据的初始最大速率。默认值没有设置。
spark.streaming.backpressure.rateEstimator：速率估算器类，默认值为 pid ，目前 Spark 只支持这个，大家可以根据自己的需要实现。
spark.streaming.backpressure.pid.proportional：用于响应错误的权重（最后批次和当前批次之间的更改）。默认值为1，只能设置成非负值。weight for response to “error” (change between last batch and this batch)
spark.streaming.backpressure.pid.integral：错误积累的响应权重，具有抑制作用（有效阻尼）。默认值为 0.2 ，只能设置成非负值。weight for the response to the accumulation of error. This has a dampening effect.
spark.streaming.backpressure.pid.derived：对错误趋势的响应权重。 这可能会引起 batch size 的波动，可以帮助快速增加/减少容量。默认值为0，只能设置成非负值。weight for the response to the trend in error. This can cause arbitrary/noise-induced fluctuations in batch size, but can also help react quickly to increased/reduced capacity.
spark.streaming.backpressure.pid.minRate：可以估算的最低费率是多少。默认值为 100，只能设置成非负值。

57.一文搞懂 Flink 的 Exactly Once 和 At Least Once

本文主要为了让你搞懂 Flink 的 Exactly Once 和 At Least Once，看完本文，你能 get 到以下知识：
介绍 CheckPoint 如何保障 Flink 任务的高可用
CheckPoint 中的状态简介
如何实现全域一致的分布式快照？
什么是 barrier？什么是 barrier 对齐？
证明了：为什么 barrier 对齐就是 Exactly Once，为什么 barrier 不对齐就是 At Least Once。
Flink 简介
有状态函数和运算符在各个元素/事件的处理中存储数据（状态数据可以修改和查询，可以自己维护，根据自己的业务场景，保存历史数据或者中间结果到状态中）
例如：
当应用程序搜索某些事件模式时，状态将存储到目前为止遇到的事件序列。
在每分钟/小时/天聚合事件时，状态保存待处理的聚合。
当在数据点流上训练机器学习模型时，状态保持模型参数的当前版本。
当需要管理历史数据时，状态允许有效访问过去发生的事件。
什么是状态？
无状态计算的例子：
比如：我们只是进行一个字符串拼接，输入 a，输出 a_666,输入b，输出 b_666输出的结果跟之前的状态没关系，符合幂等性。
幂等性：就是用户对于同一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生了副作用
有状态计算的例子：
计算 pv、uv。
输出的结果跟之前的状态有关系，不符合幂等性，访问多次，pv 会增加。

Flink 的 CheckPoint 功能简介
1.Flink CheckPoint 的存在就是为了解决 Flink 任务 failover 掉之后，能够正常恢复任务。那 CheckPoint 具体做了哪些功能，为什么任务挂掉之后，通过 CheckPoint 能使得任务恢复呢？
2.CheckPoint 是通过给程序快照的方式使得将历史某些时刻的状态保存下来，当任务挂掉之后，默认从最近一次保存的完整快照处进行恢复任务。问题来了，快照是什么鬼？能吃吗？
3.SnapShot 翻译为快照，指将程序中某些信息存一份，后期可以用来恢复。对于一个 Flink 任务来讲，快照里面到底保存着什么信息呢？
4.晦涩难懂的概念怎么办？当然用案例来代替咯，用案例让大家理解快照里面到底存什么信息。选一个大家都比较清楚的指标，app 的 pv，Flink 该怎么统计呢？
我们从 Kafka 读取到一条条的日志，从日志中解析出 app_id，然后将统计的结果放到内存中一个 Map 集合，app_id 作为 key，对应的 pv 做为 value，每次只需要将相应 app_id 的 pv 值 +1 后 put 到 Map 中即可。

Flink 任务 task 图
5.Flink 的 Source task 记录了当前消费到 kafka test topic 的所有 partition 的 offset，为了方便理解 CheckPoint 的作用，这里先用一个 partition 进行讲解，假设名为 “test”的 topic 只有一个 partition0。
例：（0，1000）
表示 0 号 partition 目前消费到 offset 为 1000 的数据
6.Flink 的 pv task 记录了当前计算的各 app 的 pv 值，为了方便讲解，我这里有两个 app：app1、app2
例：（app1，50000）（app2，10000）
表示 app1 当前 pv 值为 50000
表示 app2 当前 pv 值为 10000
每来一条数据，只需要确定相应 app_id，将相应的 value 值 +1 后 put 到 map 中即可。
7.该案例中，CheckPoint 到底记录了什么信息呢？
offset：（0，1000）
pv：（app1，50000）（app2，10000）
记录的其实就是第 n 次 CheckPoint 消费的 offset 信息和各 app 的 pv 值信息，记录一下发生 CheckPoint 当前的状态信息，并将该状态信息保存到相应的状态后端。（注：状态后端是保存状态的地方，决定状态如何保存，如何保障状态高可用，我们只需要知道，我们能从状态后端拿到 offset 信息和 pv 信息即可。状态后端必须是高可用的，否则我们的状态后端经常出现故障，会导致无法通过 checkpoint 来恢复我们的应用程序）
chk-100
该状态信息表示第 100 次 CheckPoint 的时候， partition 0 offset 消费到了 1000，pv 统计结果为（app1，50000）（app2，10000）。
8.任务挂了，如何恢复？
假如我们设置了三分钟进行一次 CheckPoint，保存了上述所说的 chk-100 的 CheckPoint 状态后，过了十秒钟，offset 已经消费到（0，1100），pv 统计结果变成了（app1，50080）（app2，10020），但是突然任务挂了，怎么办？
莫慌，其实很简单，flink只需要从最近一次成功的 CheckPoint 保存的offset（0，1000）处接着消费即可，当然pv值也要按照状态里的 pv 值（app1，50000）（app2，10000）进行累加，不能从（app1，50080）（app2，10020）处进行累加，因为 partition 0 offset 消费到 1000 时，pv 统计结果为（app1，50000）（app2，10000）。
当然如果你想从 offset （0，1100）pv（app1，50080）（app2，10020）这个状态恢复，也是做不到的，因为那个时刻程序突然挂了，这个状态根本没有保存下来。我们能做的最高效方式就是从最近一次成功的 CheckPoint 处恢复，也就是我一直所说的 chk-100。
以上讲解，基本就是 CheckPoint 承担的工作，描述的场景比较简单。
9.疑问，计算 pv 的 task 在一直运行，它怎么知道什么时候去做这个快照？或者说计算 pv 的 task 怎么保障它自己计算的 pv 值（app1，50000）（app2，10000）就是 offset（0，1000）那一刻的统计结果呢？

barrier
barrier 从 Source Task 处生成，一直流到 Sink Task，期间所有的 Task 只要碰到barrier，就会触发自身进行快照。
CheckPoint barrier n-1 处做的快照就是指 Job 从开始处理到 barrier n-1所有的状态数据。
barrier n 处做的快照就是指从 Job 开始到处理到 barrier n 所有的状态数据。
对应到 pv 案例中就是，SourceTask 接收到 JobManager 的编号为 chk-100 的 CheckPoint 触发请求后，发现自己恰好接收到 kafka offset（0，1000）处的数据，所以会往 offset（0，1000）数据之后 offset（0，1001）数据之前安插一个 barrier，然后自己开始做快照，也就是将 offset（0，1000）保存到状态后端 chk-100 中。然后 barrier 接着往下游发送，当统计 pv 的 task 接收到 barrier 后，也会暂停处理数据，将自己内存中保存的 pv 信息（app1，50000）。（app2，10000）保存到状态后端 chk-100 中。OK，Flink 大概就是通过这个原理来保存快照的。
统计 pv 的 task 接收到 barrier，就意味着 barrier 之前的数据都处理了，所以说，不会出现丢数据的情况。
barrier 的作用就是为了把数据区分开，CheckPoint 过程中有一个同步做快照的环节不能处理 barrier 之后的数据，为什么呢？
如果做快照的同时，也在处理数据，那么处理的数据可能会修改快照内容，所以先暂停处理数据，把内存中快照保存好后，再处理数据。
结合案例来讲就是，统计 pv 的 task 想对（app1，50000）（app2，10000）做快照，但是如果数据还在处理，可能快照还没保存下来，状态已经变成了（app1，50001）（app2，10001），快照就不准确了，就不能保障 Exactly Once 了。
Flink 是在数据中加了一个叫做 barrier 的东西（barrier 中文翻译：栅栏），上图中红圈处就是两个 barrier。
10.总结
流式计算中状态交互

流式计算中状态交互
11.简易场景精确一次的容错方法

checkpoint 简介 1

checkpoint 简介 2

checkpoint 简介 3
消费到 Y 位置的时候，将 Y 对应的状态保存下来
消费到 X 位置的时候，将 X 对应的状态保存下来
周期性地对消费 offset 和统计的状态信息或统计结果进行快照

多并行度、多 Operator 情况下，CheckPoint 过程

1.分布式状态容错面临的问题与挑战
如何确保状态拥有精确一次的容错保证？
如何在分布式场景下替多个拥有本地状态的算子产生一个全域一致的快照？
如何在不中断运算的前提下产生快照？

2.多并行度、多 Operator 实例的情况下，如何做全域一致的快照？
所有的 Operator 运行过程中遇到 barrier 后，都对自身的状态进行一次快照，保存到相应状态后端。

多并行度CheckPoint快照简图
对应到 pv 案例：有的 Operator 计算的 app1 的 pv，有的 Operator 计算的 app2的 pv，当他们碰到 barrier 时，都需要将目前统计的 pv 信息快照到状态后端。

3.多 Operator 状态恢复

多并行度CheckPoint恢复
4.具体怎么做这个快照呢？
利用之前所有的 barrier 策略。

barrier
JobManager 向 SourceTask 发送 CheckPointTrigger，SourceTask 会在数据流中安插 CheckPoint barrier。

多并行度快照详图 0
Source Task 自身做快照，并保存到状态后端。

多并行度快照详图 1
Source Task 将 barrier 跟数据流一块往下游发送。

多并行度快照详图 2
当下游的 Operator 实例接收到 CheckPointbarrier 后，对自身做快照。

多并行度快照详图 3

多并行度快照详图 4
上述图中，有 4 个带状态的 Operator 实例，相应的状态后端就可以想象成填 4 个格子。整个 CheckPoint 的过程可以当做 Operator 实例填自己格子的过程，Operator 实例将自身的状态写到状态后端中相应的格子，当所有的格子填满可以简单的认为一次完整的 CheckPoint 做完了。

5.上面只是快照的过程，整个 CheckPoint 执行过程如下
JobManager 端的 CheckPointCoordinator 向所有 SourceTask 发送 CheckPointTrigger，Source Task 会在数据流中安插 CheckPoint barrier。
当 task 收到所有的 barrier 后，向自己的下游继续传递 barrier，然后自身执行快照，并将自己的状态异步写入到持久化存储中。
增量 CheckPoint 只是把最新的一部分更新写入到 外部存储;
为了下游尽快做 CheckPoint，所以会先发送 barrier 到下游，自身再同步进行快照;
当 task 完成备份后，会将备份数据的地址（state handle）通知给 JobManager 的 CheckPointCoordinator。
如果 CheckPoint 的持续时长超过了 CheckPoint 设定的超时时间，CheckPointCoordinator 还没有收集完所有的 State Handle，CheckPointCoordinator 就会认为本次 CheckPoint 失败，会把这次 CheckPoint 产生的所有状态数据全部删除。
最后 CheckPointCoordinator 会把整个 StateHandle 封装成 completed CheckPoint Meta，写入到 hdfs。

6.barrier 对齐
什么是 barrier 对齐？

stream_aligning
一旦 Operator 从输入流接收到 CheckPoint barrier n，它就不能处理来自该流的任何数据记录，直到它从其他所有输入接收到 barrier n 为止。否则，它会混合属于快照 n 的记录和属于快照 n + 1 的记录。
接收到 barrier n 的流暂时被搁置。从这些流接收的记录不会被处理，而是放入输入缓冲区。
上图中第 2 个图，虽然数字流对应的 barrier 已经到达了，但是 barrier 之后的 1、2、3 这些数据只能放到 buffer 中，等待字母流的 barrier 到达。
一旦最后所有输入流都接收到 barrier n，Operator 就会把缓冲区中 pending 的输出数据发出去，然后把 CheckPoint barrier n 接着往下游发送。这里还会对自身进行快照。
之后，Operator 将继续处理来自所有输入流的记录，在处理来自流的记录之前先处理来自输入缓冲区的记录。
什么是 barrier 不对齐？
上述图 2 中，当还有其他输入流的 barrier 还没有到达时，会把已到达的 barrier 之后的数据 1、2、3 搁置在缓冲区，等待其他流的 barrier 到达后才能处理。
barrier 不对齐就是指当还有其他流的 barrier 还没到达时，为了不影响性能，也不用理会，直接处理 barrier 之后的数据。等到所有流的 barrier 的都到达后，就可以对该 Operator 做 CheckPoint 了
为什么要进行 barrier 对齐？不对齐到底行不行？
Exactly Once 时必须 barrier 对齐，如果 barrier 不对齐就变成了 At Least Once。后面的部分主要证明这句话。
CheckPoint 的目的就是为了保存快照，如果不对齐，那么在 chk-100 快照之前，已经处理了一些 chk-100 对应的 offset 之后的数据，当程序从 chk-100 恢复任务时，chk-100 对应的 offset 之后的数据还会被处理一次，所以就出现了重复消费。如果听不懂没关系，后面有案例让您懂。
结合 pv 案例来看，之前的案例为了简单，描述的 kafka 的 topic 只有 1 个 partition，这里为了讲述 barrier 对齐，所以 topic 有 2 个 partittion。

flink 消费 kafka，计算 pv 详图
Flink 同样会起四个 Operator 实例，我还称他们是 TaskA0、TaskA1、TaskB0、TaskB1。四个 Operator 会从状态后端读取保存的状态信息。
从 offset：(0，10000)(1，10005) 开始消费，并且基于 pv：(app0，8000)(app1，12050)值进行累加统计。
然后你就应该会发现这个 app1 的 pv 值 12050 实际上已经包含了 partition1 的 offset 10005~10200 的数据，所以 partition1 从 offset 10005 恢复任务时，partition1 的 offset 10005~10200 的数据被消费了两次。
TaskB1 设置的 barrier 不对齐，所以 CheckPoint chk-100 对应的状态中多消费了 barrier 之后的一些数据（TaskA1 发送），重启后是从 chk-100 保存的 offset 恢复，这就是所说的 At Least Once。
由于上面说 TaskB0 设置的 barrier 对齐，所以 app0 不会出现重复消费，因为 app0 没有消费 offset：(0，10000)(1，10005) 之后的数据，也就是所谓的 Exactly Once。
chk-100
offset：(0，10000)(1，10005)
pv：(app0，8000) (app1，12050)
虽然状态保存的 pv 值偏高了，但是不能说明重复处理，因为我的 TaskA1 并没有再次去消费 partition1 的 offset 10005~10200 的数据，所以相当于也没有重复消费，只是展示的结果更实时了。
这里假如 TaskA0 消费的 partition0 的 offset 为 10000，TaskA1 消费的 partition1 的 offset 为 10005。那么状态中会保存 (0，10000)(1，10005)，表示 0 号 partition 消费到了 offset 为 10000 的位置，1 号 partition 消费到了 offset 为 10005 的位置。
结合业务，先介绍一下上述所有算子在业务中的功能：
Source 的 kafka 的 Consumer，从 kakfa 中读取数据到 Flink 应用中
TaskA 中的 map 将读取到的一条 kafka 日志转换为我们需要统计的 app_id
keyBy 按照 app_id 进行 keyBy，相同的 app_id 会分到下游 TaskB的同一个实例中
TaskB 的 map 在状态中查出该 app_id 对应的 pv 值，然后 +1，存储到状态中
利用 Sink 将统计的 pv 值写入到外部存储介质中
我们从 kafka 的两个 partition 消费数据，TaskA 和 TaskB 都有两个并行度，所以总共 Flink 有 4 个 Operator 实例，这里我们称之为 TaskA0、TaskA1、TaskB0、TaskB1。
假设已经成功做了 99 次 CheckPoint，这里详细解释第 100 次 CheckPoint 过程。
JobManager 内部有个定时调度，假如现在 10 点 00 分 00 秒到了第 100 次 CheckPoint 的时间了，JobManager 的 CheckPointCoordinator 进程会向所有的 Source Task 发送 CheckPointTrigger，也就是向 TaskA0、TaskA1 发送 CheckPointTrigger。
TaskA0、TaskA1 接收到 CheckPointTrigger，会往数据流中安插 barrier，将 barrier 发送到下游，在自己的状态中记录 barrier 安插的 offset 位置，然后自身做快照，将 offset 信息保存到状态后端。然后 TaskA 的 map 和 keyBy 算子中并没有状态，所以不需要进行快照。
接着数据和 barrier 都向下游 TaskB 发送，相同的 app_id 会发送到相同的TaskB实例上，这里假设有两个 app：app0 和 app1，经过 keyBy 后，假设 app0 分到了 TaskB0 上，app1 分到了 TaskB1 上。基于上面描述，TaskA0 和 TaskA1 中的所有 app0 的数据都发送到 TaskB0 上，所有 app1 的数据都发送到 TaskB1 上。
现在我们假设 TaskB0 做 CheckPoint 的时候 barrier 对齐了，TaskB1 做 CheckPoint 的时候 barrier 不对齐，当然不能这么配置，我就是举这么个例子，带大家分析一下 barrier 对不对齐到底对统计结果有什么影响？
上面说了 chk-100 的这次 CheckPoint，offset 位置为(0，10000)(1，10005)，TaskB0 使用 barrier 对齐，也就是说 TaskB0 不会处理 barrier 之后的数据，所以TaskB0 在 chk-100 快照的时候，状态后端保存的 app0 的 pv 数据是从程序开始启动到 kafkaoffset 位置为(0，10000)(1，10005)的所有数据计算出来的 pv 值，一条不多（没处理 barrier 之后，所以不会重复），一条不少(barrier 之前的所有数据都处理了，所以不会丢失)，假如保存的状态信息为(app0，8000)表示消费到(0，10000)(1，10005)offset 的时候，app0 的 pv 值为 8000。
TaskB1 使用的 barrier 不对齐，假如 TaskA0 由于服务器的 CPU 或者网络等其他波动，导致 TaskA0 处理数据较慢，而 TaskA1 很稳定，所以处理数据比较快。导致的结果就是 TaskB1 先接收到了 TaskA1 的 barrier，由于配置的 barrier 不对齐，所以 TaskB1 会接着处理 TaskA1 barrier 之后的数据，过了 2 秒后，TaskB1 接收到了 TaskA0 的 barrier，于是对状态中存储的 app1 的 pv 值开始做 CheckPoint 快照，保存的状态信息为(app1，12050)，但是我们知道这个(app1，12050)实际上多处理了 2 秒 TaskA1 发来的 barrier 之后的数据，也就是 kafka topic 对应的 partition1 offset 10005 之后的数据，app1 真实的 pv 数据肯定要小于这个 12050，partition1 的 offset 保存的 offset 虽然是 10005，但是我们实际上可能已经处理到了 offset 10200 的数据，假设就是处理到了 10200。
分析到这里，我们先梳理一下我们的状态保存了什么：
chk-100
offset：(0，10000)(1，10005)
pv：(app0，8000) (app1，12050)
接着程序在继续运行，过了 10 秒，由于某个服务器挂了，导致我们的四个 Operator 实例有一个 Operator 挂了，所以 Flink 会从最近一次的状态恢复，也就是我们刚刚详细讲的 chk-100 处恢复，那具体是怎么恢复的呢？
Flink 同样会起四个 Operator 实例，我还称他们是 TaskA0、TaskA1、TaskB0、TaskB1。四个 Operator 会从状态后端读取保存的状态信息。
从 offset：(0，10000)(1，10005) 开始消费，并且基于 pv：(app0，8000) (app1，12050)值进行累加统计
然后你就应该会发现这个 app1 的 pv 值 12050 实际上已经包含了 partition1 的 offset 10005~10200 的数据，所以 partition1 从 offset 10005 恢复任务时，partition1 的 offset 10005~10200 的数据被消费了两次。
TaskB1 设置的 barrier 不对齐，所以 CheckPoint chk-100 对应的状态中多消费了 barrier 之后的一些数据（TaskA1 发送），重启后是从 chk-100 保存的 offset 恢复，这就是所说的 At Least Once。
由于上面说 TaskB0 设置的 barrier 对齐，所以 app0 不会出现重复消费，因为 app0 没有消费 offset：(0，10000)(1，10005) 之后的数据，也就是所谓的 Exactly Once。
看到这里你应该已经知道了哪种情况会出现重复消费了，也应该要掌握为什么 barrier 对齐就是 Exactly Once，为什么 barrier 不对齐就是 At Least Once。
这里再补充一个问题，到底什么时候会出现 barrier 对齐？
首先设置了 Flink 的 CheckPoint 语义是：Exactly Once。
Operator 实例必须有多个输入流才会出现 barrier 对齐。
对齐，汉语词汇，释义为使两个以上事物配合或接触得整齐。由汉语解释可得对齐肯定需要两个以上事物，所以，必须有多个流才叫对齐。barrier 对齐其实也就是上游多个流配合使得数据对齐的过程。
言外之意：如果 Operator 实例只有一个输入流，就根本不存在 barrier 对齐，自己跟自己默认永远都是对齐的。
Q & A 环节
第一种场景计算 PV，kafka 只有一个 partition，精确一次，至少一次就没有区别？
答：如果只有一个 partition，对应 Flink 任务的 Source Task 并行度只能是 1，确实没有区别，不会有至少一次的存在了，肯定是精确一次。因为只有 barrier 不对齐才会有可能重复处理，这里并行度都已经为 1，默认就是对齐的，只有当上游有多个并行度的时候，多个并行度发到下游的 barrier 才需要对齐，单并行度不会出现 barrier 不对齐，所以必然精确一次。其实还是要理解 barrier 对齐就是 Exactly Once 不会重复消费，barrier 不对齐就是 At Least Once 可能重复消费，这里只有单个并行度根本不会存在 barrier 不对齐，所以不会存在至少一次语义。
为了下游尽快做 CheckPoint，所以会先发送 barrier 到下游，自身再同步进行快照；这一步，如果向下发送 barrier 后，自己同步快照慢怎么办？下游已经同步好了，自己还没？
答: 可能会出现下游比上游快照还早的情况，但是这不影响快照结果，只是下游快照的更及时了，我只要保障下游把 barrier 之前的数据都处理了，并且不处理 barrier 之后的数据，然后做快照，那么下游也同样支持精确一次。这个问题你不要从全局思考，你单独思考上游和下游的实例，你会发现上下游的状态都是准确的，既没有丢，也没有重复计算。
这里需要注意一点，如果有一个 Operator 的 CheckPoint 失败了或者因为 CheckPoint 超时也会导致失败，那么 JobManager 会认为整个 CheckPoint 失败。失败的 CheckPoint 是不能用来恢复任务的，必须所有的算子的 CheckPoint 都成功，那么这次 CheckPoint 才能认为是成功的，才能用来恢复任务。
我程序中 Flink 的 CheckPoint 语义设置了 Exactly Once，但是我的 MySQL 中看到数据重复了？程序中设置了 1 分钟 1 次 CheckPoint，但是 5 秒向 MySQL 写一次数据，并 commit。
答：Flink 要求 end to end 的精确一次都必须实现TwoPhaseCommitSinkFunction。如果你的 chk-100 成功了，过了 30 秒，由于 5 秒 commit 一次，所以实际上已经写入了 6 批数据进入 MySQL，但是突然程序挂了，从 chk100 处恢复，这样的话，之前提交的 6 批数据就会重复写入，所以出现了重复消费。Flink 的精确一次有两种情况，一个是 Flink 内部的精确一次，一个是端对端的精确一次，这个博客所描述的都是关于 Flink 内部去的精确一次，我后期再发一个博客详细介绍一下 Flink 端对端的精确一次如何实现。
这篇文章有这么一句话 TwoPhaseCommitSinkFunction 已经把这种情况考虑在内了，并且在从 checkpoint 点恢复状态时，会优先发出一个 commit。个人感觉只要把这句话理解了，知道为什么每次恢复状态时，都需要优先发出一个 commit，那就把 Flink 的 TwoPhaseCommitSinkFunction 真正理解了。

58.谈谈流计算中的『Exactly Once』特性

背景
流处理（有时称为事件处理）可以简单地描述为是对*数据或事件的连续处理。流或事件处理应用程序可以或多或少地被描述为有向图，并且通常被描述为有向无环图（DAG）。在这样的图中，每个边表示数据或事件流，每个顶点表示运算符，会使用程序中定义的逻辑处理来自相邻边的数据或事件。有两种特殊类型的顶点，通常称为 sources 和 sinks。sources读取外部数据/事件到应用程序中，而 sinks 通常会收集应用程序生成的结果。下图是流式应用程序的示例。

A typical stream processing topology
流处理引擎通常允许用户指定可靠性模式或处理语义，以指示它将为整个应用程序中的数据处理提供哪些保证。这些保证是有意义的，因为你始终会遇到由于网络，机器等可能导致数据丢失的故障。流处理引擎通常为应用程序提供了三种数据处理语义：最多一次、至少一次和精确一次。
如下是对这些不同处理语义的宽松定义：
最多一次（At-most-once）
这本质上是一『尽力而为』的方法。保证数据或事件最多由应用程序中的所有算子处理一次。 这意味着如果数据在被流应用程序完全处理之前发生丢失，则不会进行其他重试或者重新发送。下图中的例子说明了这种情况。

At-most-once processing semantics
至少一次（At-least-once）
应用程序中的所有算子都保证数据或事件至少被处理一次。这通常意味着如果事件在流应用程序完全处理之前丢失，则将从源头重放或重新传输事件。然而，由于事件是可以被重传的，因此一个事件有时会被处理多次，这就是所谓的至少一次。
下图的例子描述了这种情况：第一个算子最初未能成功处理事件，然后在重试时成功，接着在第二次重试时也成功了，其实是没有必要的。

At-least-once processing semantics
精确一次（Exactly-once）
即使是在各种故障的情况下，流应用程序中的所有算子都保证事件只会被『精确一次』的处理。（也有文章将 Exactly-once 翻译为：完全一次，恰好一次）
通常使用两种流行的机制来实现『精确一次』处理语义。
– 分布式快照 / 状态检查点
– 至少一次事件传递和对重复数据去重
实现『精确一次』的分布式快照/状态检查点方法受到 Chandy-Lamport 分布式快照算法的启发[1]。通过这种机制，流应用程序中每个算子的所有状态都会定期做 checkpoint。如果是在系统中的任何地方发生失败，每个算子的所有状态都回滚到最新的全局一致 checkpoint 点。在回滚期间，将暂停所有处理。源也会重置为与最近 checkpoint 相对应的正确偏移量。整个流应用程序基本上是回到最近一次的一致状态，然后程序可以从该状态重新启动。下图描述了这种 checkpoint 机制的基础知识。

Distributed snapshot
在上图中，流应用程序在 T1 时间处正常工作，并且做了checkpoint。然而，在时间 T2，算子未能处理输入的数据。此时，S=4 的状态值已保存到持久存储器中，而状态值 S=12 保存在算子的内存中。为了修复这种差异，在时间 T3，处理程序将状态回滚到 S=4 并“重放”流中的每个连续状态直到最近，并处理每个数据。最终结果是有些数据已被处理了多次，但这没关系，因为无论执行了多少次回滚，结果状态都是相同的。
另一种实现『精确一次』的方法是：在每个算子上实现至少一次事件传递和对重复数据去重来。使用此方法的流处理引擎将重放失败事件，以便在事件进入算子中的用户定义逻辑之前，进一步尝试处理并移除每个算子的重复事件。此机制要求为每个算子维护一个事务日志，以跟踪它已处理的事件。利用这种机制的引擎有 Google 的 MillWheel[2] 和 Apache Kafka Streams。下图说明了这种机制的要点。

At-least-once delivery plus deduplication
『精确一次』是真正的『精确一次』吗?
现在让我们重新审视『精确一次』处理语义真正对最终用户的保证。『精确一次』这个术语在描述正好处理一次时会让人产生误导。
有些人可能认为『精确一次』描述了事件处理的保证，其中流中的每个事件只被处理一次。实际上，没有引擎能够保证正好只处理一次。在面对任意故障时，不可能保证每个算子中的用户定义逻辑在每个事件中只执行一次，因为用户代码被部分执行的可能性是永远存在的。
考虑具有流处理运算符的场景，该运算符执行打印传入事件的 ID 的映射操作，然后返回事件不变。下面的伪代码说明了这个操作：
Map (Event event) {
Print "Event ID: " + event.getId()
Return event
}
每个事件都有一个 GUID (全局惟一ID)。如果用户逻辑的精确执行一次得到保证，那么事件 ID 将只输出一次。但是，这是无法保证的，因为在用户定义的逻辑的执行过程中，随时都可能发生故障。引擎无法自行确定执行用户定义的处理逻辑的时间点。因此，不能保证任意用户定义的逻辑只执行一次。这也意味着，在用户定义的逻辑中实现的外部操作(如写数据库)也不能保证只执行一次。此类操作仍然需要以幂等的方式执行。
那么，当引擎声明『精确一次』处理语义时，它们能保证什么呢？如果不能保证用户逻辑只执行一次，那么什么逻辑只执行一次？当引擎声明『精确一次』处理语义时，它们实际上是在说，它们可以保证引擎管理的状态更新只提交一次到持久的后端存储。
上面描述的两种机制都使用持久的后端存储作为真实性的来源，可以保存每个算子的状态并自动向其提交更新。对于机制 1 (分布式快照 / 状态检查点)，此持久后端状态用于保存流应用程序的全局一致状态检查点(每个算子的检查点状态)。对于机制 2 (至少一次事件传递加上重复数据删除)，持久后端状态用于存储每个算子的状态以及每个算子的事务日志，该日志跟踪它已经完全处理的所有事件。
提交状态或对作为真实来源的持久后端应用更新可以被描述为恰好发生一次。然而，如上所述，计算状态的更新 / 更改，即处理在事件上执行任意用户定义逻辑的事件，如果发生故障，则可能不止一次地发生。换句话说，事件的处理可以发生多次，但是该处理的效果只在持久后端状态存储中反映一次。因此，我们认为有效地描述这些处理语义最好的术语是『有效一次』（effectively once）。
分布式快照与至少一次事件传递和重复数据删除的比较
从语义的角度来看，分布式快照和至少一次事件传递以及重复数据删除机制都提供了相同的保证。然而，由于两种机制之间的实现差异，存在显着的性能差异。
机制 1（分布式快照 / 状态检查点）的性能开销是最小的，因为引擎实际上是往流应用程序中的所有算子一起发送常规事件和特殊事件，而状态检查点可以在后台异步执行。但是，对于大型流应用程序，故障可能会更频繁地发生，导致引擎需要暂停应用程序并回滚所有算子的状态，这反过来又会影响性能。流式应用程序越大，故障发生的可能性就越大，因此也越频繁，反过来，流式应用程序的性能受到的影响也就越大。然而，这种机制是非侵入性的，运行时需要的额外资源影响很小。
机制 2（至少一次事件传递加重复数据删除）可能需要更多资源，尤其是存储。使用此机制，引擎需要能够跟踪每个算子实例已完全处理的每个元组，以执行重复数据删除，以及为每个事件执行重复数据删除本身。这意味着需要跟踪大量的数据，尤其是在流应用程序很大或者有许多应用程序在运行的情况下。执行重复数据删除的每个算子上的每个事件都会产生性能开销。但是，使用这种机制，流应用程序的性能不太可能受到应用程序大小的影响。对于机制 1，如果任何算子发生故障，则需要发生全局暂停和状态回滚；对于机制 2，失败的影响更加局部性。当在算子中发生故障时，可能尚未完全处理的事件仅从上游源重放/重传。性能影响与流应用程序中发生故障的位置是隔离的，并且对流应用程序中其他算子的性能几乎没有影响。从性能角度来看，这两种机制的优缺点如下。
分布式快照 / 状态检查点的优缺点：
优点：
– 较小的性能和资源开
缺点：
– 对性能的影响较大
– 拓扑越大，对性能的潜在影响越大
至少一次事件传递以及重复数据删除机制的优缺点：
优点：
– 故障对性能的影响是局部的
– 故障的影响不一定会随着拓扑的大小而增加
缺点：
– 可能需要大量的存储和基础设施来支持
– 每个算子的每个事件的性能开销
虽然从理论上讲，分布式快照和至少一次事件传递加重复数据删除机制之间存在差异，但两者都可以简化为至少一次处理加幂等性。对于这两种机制，当发生故障时(至少实现一次)，事件将被重放/重传，并且通过状态回滚或事件重复数据删除，算子在更新内部管理状态时本质上是幂等的。
结论
在这篇博客文章中，我希望能够让你相信『精确一次』这个词是非常具有误导性的。提供『精确一次』的处理语义实际上意味着流处理引擎管理的算子状态的不同更新只反映一次。『精确一次』并不能保证事件的处理，即任意用户定义逻辑的执行，只会发生一次。我们更喜欢用『有效一次』（effectively once）这个术语来表示这种保证，因为处理不一定保证只发生一次，但是对引擎管理的状态的影响只反映一次。两种流行的机制，分布式快照和重复数据删除，被用来实现精确/有效的一次性处理语义。这两种机制为消息处理和状态更新提供了相同的语义保证，但是在性能上存在差异。这篇文章并不是要让你相信任何一种机制都优于另一种，因为它们各有利弊。

59.Time 深度解析

前言
Flink 的 API 大体上可以划分为三个层次：处于最底层的 ProcessFunction、中间一层的 DataStream API 和最上层的 SQL/Table API，这三层中的每一层都非常依赖于时间属性。时间属性是流处理中最重要的一个方面，是流处理系统的基石之一，贯穿这三层 API。在 DataStream API 这一层中因为封装方面的原因，我们能够接触到时间的地方不是很多，所以我们将重点放在底层的 ProcessFunction 和最上层的 SQL/Table API。

Flink 时间语义
在不同的应用场景中时间语义是各不相同的，Flink 作为一个先进的分布式流处理引擎，它本身支持不同的时间语义。其核心是 Processing Time 和 Event Time（Row Time），这两类时间主要的不同点如下表所示：

Processing Time 是来模拟我们真实世界的时间，其实就算是处理数据的节点本地时间也不一定就是完完全全的我们真实世界的时间，所以说它是用来模拟真实世界的时间。而 Event Time 是数据世界的时间，就是我们要处理的数据流世界里面的时间。关于他们的获取方式，Process Time 是通过直接去调用本地机器的时间，而 Event Time 则是根据每一条处理记录所携带的时间戳来判定。
这两种时间在 Flink 内部的处理以及还是用户的实际使用方面，难易程度都是不同的。相对而言的 Processing Time 处理起来更加的简单，而 Event Time 要更麻烦一些。而在使用 Processing Time 的时候，我们得到的处理结果（或者说流处理应用的内部状态）是不确定的。而因为在 Flink 内部对 Event Time 做了各种保障，使用 Event Time 的情况下，无论重放数据多少次，都能得到一个相对确定可重现的结果。
因此在判断应该使用 Processing Time 还是 Event Time 的时候，可以遵循一个原则：当你的应用遇到某些问题要从上一个 checkpoint 或者 savepoint 进行重放，是不是希望结果完全相同。如果希望结果完全相同，就只能用 Event Time；如果接受结果不同，则可以用 Processing Time。Processing Time 的一个常见的用途是，我们要根据现实时间来统计整个系统的吞吐，比如要计算现实时间一个小时处理了多少条数据，这种情况只能使用 Processing Time。

时间的特性
时间的一个重要特性是：时间只能递增，不会来回穿越。 在使用时间的时候我们要充分利用这个特性。假设我们有这么一些记录，然后我们来分别看一下 Processing Time 还有 Event Time 对于时间的处理。
对于 Processing Time，因为我们是使用的是本地节点的时间（假设这个节点的时钟同步没有问题），我们每一次取到的 Processing Time 肯定都是递增的，递增就代表着有序，所以说我们相当于拿到的是一个有序的数据流。
而在用 Event Time 的时候因为时间是绑定在每一条的记录上的，由于网络延迟、程序内部逻辑、或者其他一些分布式系统的原因，数据的时间可能会存在一定程度的乱序，比如上图的例子。在 Event Time 场景下，我们把每一个记录所包含的时间称作 Record Timestamp。如果 Record Timestamp 所得到的时间序列存在乱序，我们就需要去处理这种情况。

如果单条数据之间是乱序，我们就考虑对于整个序列进行更大程度的离散化。简单地讲，就是把数据按照一定的条数组成一些小批次，但这里的小批次并不是攒够多少条就要去处理，而是为了对他们进行时间上的划分。经过这种更高层次的离散化之后，我们会发现最右边方框里的时间就是一定会小于中间方框里的时间，中间框里的时间也一定会小于最左边方框里的时间。

这个时候我们在整个时间序列里插入一些类似于标志位的一些特殊的处理数据，这些特殊的处理数据叫做 watermark。一个 watermark 本质上就代表了这个 watermark 所包含的 timestamp 数值，表示以后到来的数据已经再也没有小于或等于这个时间的了。
Timestamp 和 Watermark 行为概览

接下来我们重点看一下 Event Time 里的 Record Timestamp（简写成 timestamp）和 watermark 的一些基本信息。绝大多数的分布式流计算引擎对于数据都是进行了 DAG 图的抽象，它有自己的数据源，有处理算子，还有一些数据汇。数据在不同的逻辑算子之间进行流动。watermark 和 timestamp 有自己的生命周期，接下来我会从 watermark 和 timestamp 的产生、他们在不同的节点之间的传播、以及在每一个节点上的处理，这三个方面来展开介绍。
Timestamp 分配和 Watermark 生成
Flink 支持两种 watermark 生成方式。第一种是在 SourceFunction 中产生，相当于把整个的 timestamp 分配和 watermark 生成的逻辑放在流处理应用的源头。我们可以在 SourceFunction 里面通过这两个方法产生 watermark：
通过 collectWithTimestamp 方法发送一条数据，其中第一个参数就是我们要发送的数据，第二个参数就是这个数据所对应的时间戳；也可以调用 emitWatermark 方法去产生一条 watermark，表示接下来不会再有时间戳小于等于这个数值记录。
另外，有时候我们不想在 SourceFunction 里生成 timestamp 或者 watermark，或者说使用的 SourceFunction 本身不支持，我们还可以在使用 DataStream API 的时候指定，调用的 DataStream.assignTimestampsAndWatermarks 这个方法，能够接收不同的 timestamp 和 watermark 的生成器。
总体上而言生成器可以分为两类：第一类是定期生成器；第二类是根据一些在流处理数据流中遇到的一些特殊记录生成的。

两者的区别主要有三个方面，首先定期生成是现实时间驱动的，这里的“定期生成”主要是指 watermark（因为 timestamp 是每一条数据都需要有的），即定期会调用生成逻辑去产生一个 watermark。而根据特殊记录生成是数据驱动的，即是否生成 watermark 不是由现实时间来决定，而是当看到一些特殊的记录就表示接下来可能不会有符合条件的数据再发过来了，这个时候相当于每一次分配 Timestamp 之后都会调用用户实现的 watermark 生成方法，用户需要在生成方法中去实现 watermark 的生成逻辑。
大家要注意的是就是我们在分配 timestamp 和生成 watermark 的过程，虽然在 SourceFunction 和 DataStream 中都可以指定，但是还是建议生成的工作越靠近 DataSource 越好。这样会方便让程序逻辑里面更多的 operator 去判断某些数据是否乱序。Flink 内部提供了很好的机制去保证这些 timestamp 和 watermark 被正确地传递到下游的节点。
Watermark 传播

具体的传播策略基本上遵循这三点。
首先，watermark 会以广播的形式在算子之间进行传播。比如说上游的算子，它连接了三个下游的任务，它会把自己当前的收到的 watermark 以广播的形式传到下游。
第二，如果在程序里面收到了一个 Long.MAX_VALUE 这个数值的 watermark，就表示对应的那一条流的一个部分不会再有数据发过来了，它相当于就是一个终止的一个标志。
第三，对于单流而言，这个策略比较好理解，而对于有多个输入的算子，watermark 的计算就有讲究了，一个原则是：单输入取其大，多输入取小。
举个例子，假设这边蓝色的块代表一个算子的一个任务，然后它有三个输入，分别是 W1、W2、W3，这三个输入可以理解成任何的输入，这三个输入可能是属于同一个流，也可能是属于不同的流。然后在计算 watermark 的时候，对于单个输入而言是取他们的最大值，因为我们都知道 watermark 应该遵循一个单调递增的一个原则。对于多输入，它要统计整个算子任务的 watermark 时，就会取这三个计算出来的 watermark 的最小值。即一个多个输入的任务，它的 watermark 受制于最慢的那条输入流。这一点类似于木桶效应，整个木桶中装的水会就是受制于最矮的那块板。
watermark 在传播的时候有一个特点是，它的传播是幂等的。多次收到相同的 watermark，甚至收到之前的 watermark 都不会对最后的数值产生影响，因为对于单个输入永远是取最大的，而对于整个任务永远是取一个最小的。
同时我们可以注意到这种设计其实有一个局限，具体体现在它没有区分你这个输入是一条流多个 partition 还是来自于不同的逻辑上的流的 JOIN。对于同一个流的不同 partition，我们对他做这种强制的时钟同步是没有问题的，因为一开始就是把一条流拆散成不同的部分，但每一个部分之间共享相同的时钟。但是如果算子的任务是在做类似于 JOIN 操作，那么要求你两个输入的时钟强制同步其实没有什么道理的，因为完全有可能是把一条离现在时间很近的数据流和一个离当前时间很远的数据流进行 JOIN，这个时候对于快的那条流，因为它要等慢的那条流，所以说它可能就要在状态中去缓存非常多的数据，这对于整个集群来说是一个很大的性能开销。
ProcessFunction
在正式介绍 watermark 的处理之前，先简单介绍 ProcessFunction，因为 watermark 在任务里的处理逻辑分为内部逻辑和外部逻辑。外部逻辑其实就是通过 ProcessFunction 来体现的，如果你需要使用 Flink 提供的时间相关的 API 的话就只能写在 ProcessFunction 里。
ProcessFunction 和时间相关的功能主要有三点：
第一点就是根据你当前系统使用的时间语义不同，你可以去获取当前你正在处理这条记录的 Record Timestamp，或者当前的 Processing Time。
第二点就是它可以获取当前算子的时间，可以把它理解成当前的 watermark。
第三点就是为了在 ProcessFunction 中去实现一些相对复杂的功能，允许注册一些 timer（定时器）。比如说在 watermark 达到某一个时间点的时候就触发定时器，所有的这些回调逻辑也都是由用户来提供，涉及到如下三个方法，registerEventTimeTimer、registerProcessingTimeTimer 和 onTimer。在 onTimer 方法中就需要去实现自己的回调逻辑，当条件满足时回调逻辑就会被触发。
一个简单的应用是，我们在做一些时间相关的处理的时候，可能需要缓存一部分数据，但这些数据不能一直去缓存下去，所以需要有一些过期的机制，我们可以通过 timer 去设定这么一个时间，指定某一些数据可能在将来的某一个时间点过期，从而把它从状态里删除掉。所有的这些和时间相关的逻辑在 Flink 内部都是由自己的 Time Service（时间服务）完成的。
Watermark 处理

一个算子的实例在收到 watermark 的时候，首先要更新当前的算子时间，这样的话在 ProcessFunction 里方法查询这个算子时间的时候，就能获取到最新的时间。第二步它会遍历计时器队列，这个计时器队列就是我们刚刚说到的 timer，你可以同时注册很多 timer，Flink 会把这些 Timer 按照触发时间放到一个优先队列中。第三步 Flink 得到一个时间之后就会遍历计时器的队列，然后逐一触发用户的回调逻辑。 通过这种方式，Flink 的某一个任务就会将当前的 watermark 发送到下游的其他任务实例上，从而完成整个 watermark 的传播，从而形成一个闭环。
Table API 中的时间
下面我们来看一看 Table/SQL API 中的时间。为了让时间参与到 Table/SQL 这一层的运算中，我们需要提前把时间属性放到表的 schema 中，这样的话我们才能够在 SQL 语句或者 Table 的一些逻辑表达式里面去使用这些时间去完成需求。
Table 中指定时间列
其实之前社区就怎么在 Table/SQL 中去使用时间这个问题做过一定的讨论，是把获取当前 Processing Time 的方法是作为一个特殊的 UDF，还是把这一个列物化到整个的 schema 里面，最终采用了后者。我们这里就分开来讲一讲 Processing Time 和 Event Time 在使用的时候怎么在 Table 中指定。

对于 Processing Time，我们知道要得到一个 Table 对象（或者注册一个 Table）有两种手段：
可以从一个 DataStream 转化成一个 Table；
直接通过 TableSource 去生成这么一个 Table；
对于第一种方法而言，我们只需要在你已有的这些列中（例子中 f1 和 f2 就是两个已有的列），在最后用“列名.proctime”这种写法就可以把最后的这一列注册为一个 Processing Time，以后在写查询的时候就可以去直接使用这一列。如果 Table 是通过 TableSource 生成的，就可以通过实现这一个 DefinedRowtimeAttributes 接口，然后就会自动根据你提供的逻辑去生成对应的 Processing Time。
相对而言，在使用 Event Time 时则有一个限制，因为 Event Time 不像 Processing Time 那样是随拿随用。如果你要从 DataStream 去转化得到一个 Table，必须要提前保证原始的 DataStream 里面已经存在了 Record Timestamp 和 watermark。如果你想通过 TableSource 生成的，也一定要保证你要接入的一个数据里面存在一个类型为 long 或者 timestamp 的这么一个时间字段。
具体来说，如果你要从 DataStream 去注册一个表，和 proctime 类似，你只需要加上“列名.rowtime”就可以。需要注意的是，如果你要用 Processing Time，必须保证你要新加的字段是整个 schema 中的最后一个字段，而 Event Time 的时候你其实可以去替换某一个已有的列，然后 Flink 会自动的把这一列转化成需要的 rowtime 这个类型。 如果是通过 TableSource 生成的，只需要实现 DefinedRowtimeAttributes 接口就可以了。需要说明的一点是，在 DataStream API 这一侧其实不支持同时存在多个 Event Time（rowtime），但是在 Table 这一层理论上可以同时存在多个 rowtime。因为 DefinedRowtimeAttributes 接口的返回值是一个对于 rowtime 描述的 List，即其实可以同时存在多个 rowtime 列，在将来可能会进行一些其他的改进，或者基于去做一些相应的优化。
时间列和 Table 操作

指定完了时间列之后，当我们要真正去查询时就会涉及到一些具体的操作。这里我列举的这些操作都是和时间列紧密相关，或者说必须在这个时间列上才能进行的。比如说“Over 窗口聚合”和“Group by 窗口聚合”这两种窗口聚合，在写 SQL 提供参数的时候只能允许你在这个时间列上进行这种聚合。第三个就是时间窗口聚合，你在写条件的时候只支持对应的时间列。最后就是排序，我们知道在一个无尽的数据流上对数据做排序几乎是不可能的事情，但因为这个数据本身到来的顺序已经是按照时间属性来进行排序，所以说我们如果要对一个 DataStream 转化成 Table 进行排序的话，你只能是按照时间列进行排序，当然同时你也可以指定一些其他的列，但是时间列这个是必须的，并且必须放在第一位。
为什么说这些操作只能在时间列上进行？因为我们有的时候可以把到来的数据流就看成是一张按照时间排列好的一张表，而我们任何对于表的操作，其实都是必须在对它进行一次顺序扫描的前提下完成的。因为大家都知道数据流的特性之一就是一过性，某一条数据处理过去之后，将来其实不太好去访问它。当然因为 Flink 中内部提供了一些状态机制，我们可以在一定程度上去弱化这个特性，但是最终还是不能超越的限制状态不能太大。所有这些操作为什么只能在时间列上进行，因为这个时间列能够保证我们内部产生的状态不会无限的增长下去，这是一个最终的前提。

60.Flink Kafka 端到端Exactly-Once语义

61.Apache Flink 结合 Kafka 构建端到端的 Exactly-Once 处理

61.Flink Kafka Connector 与 Exactly-Once剖析