流处理系统(Flink, Kafka和Pravega)学习笔记

本文记录一下我比较感兴趣的实时流处理方面的知识，从计算Flink，到存储Pravega，再到消息中间件Kafka，理论知识四大方面学习。如有错误请在评论区指正。实时更新~

理论知识

有状态的流计算

无状态的意思是流处理的时候只需要针对某一条消息进行处理，结果只受到这条消息的影响，比如在每一条消息后面追加字符“a”；

有状态指的是在消息处理的时候需要保存前后多条消息的相关信息，结果受到多条消息的影响，比如count，average操作。所以有状态操作更加强大但是实现起来更加困难，特别是当它也支持“恰好一次”的时候。

流消息的三种语义

At most Once

接收者最多收到一次消息

At Least Once

发送者给接收者发送消息，如果一直收不到接收者的确诊消息，发送者会一直重发。

Exactly Once

对于一条消息，接收者确保只收到一次。每个输入的事件最终只会影响结果1次，即使机器或者软件出现故障，既没有重复数据，也不会丢数据。

不管处理时候有没有错误发生，计算的结果都应该是一样的。

所以在计算的时候如果发生了错误，系统重新计算，结果一定要和没有错误发生的结果是一样的。

先讲讲其他流处理架构是如何做fault tolerance的.

1.Storm的Record acknowledgement

每一个被storm的operator处理的数据都会向其上一个operator发送一份应答消息，通知其已被下游处理。storm的源operator保存了所有已发送的消息的每一个下游算子的应答消息，当它收到来自sink的应答时，它就知道该消息已经被完整处理，可以移除了。
如果没有收到应答，storm就会重发该消息。显而易见，这是一种at least once的逻辑。另外，这种方式面临着严重的幂等性问题，例如对一个count算子，如果count的下游算子出错，source重发该消息，那么防止该消息被count两遍的逻辑需要程序员自己去实现

2.Spark streaming的micro batch

通过控制每批计算数据的大小来控制延迟与吞吐量的制约，如果想要低延迟，就用小一点的batch，如果想要大吞吐量，就不得不忍受更高的延迟

我们重点来看Flink与Kafka的exactly-once语义支持

Flink 对Exactly Once的支持

实现Exactly-Once的关键在于能够准确知道和快速记录下来当前的operator的状态

Flink定期会通过记录checkpoint，即一下信息的的一致性快照。

应用程序的当前状态
输入流的位置

Flink可以配置一个固定的时间点，定期产生checkpoint，然后把checkpoint的数据写入持久存储系统。如S3或HDFS。将checkpoint数据写入持久存储是异步发生的，这意味着Flink应用程序在checkpoint过程中可以继续处理数据。

发生机器或软件故障，重新启动后，Flink应用程序将从最新的checkpoint点恢复处理； Flink会恢复应用程序状态，将输入流回滚到上次checkpoint保存的位置，然后重新开始运行。这意味着Flink可以像从未发生过故障一样计算结果。

但以上的exactly-once语义都是基于flink内部实现的

那如果要实现端到端的Exactly-Once语义，即Flink写入外部系统也需要能够满足Exactly-Once语义。这些外部系统必须要提供提交或者回滚的方法，然后通过Flink的checkpoint来维护一致性。

因此就要引入两阶段提交协议这个在分布式系统中协调提交和回滚的方法了。

然后下面讲一下Kafka的exactly-once

如果Kafka不支持exactly-once操作，那么可能出现下面的错误：

1.重复写入

2.计算状态多次更新

3.重复消费

那Kafka是怎么保证自己的exactly-once呢？

1.把计算结果写入输出的topic中——生产者提交数据到broker
2.broker把更新操作写入更新日志changelog中——broker进行消息处理
3.消费者消费数据，提交偏移量，把消息的消息偏移量写入相应的topic中

对于重复写入问题：

Kafka是怎么实现消息传输的幂等操作呢?每一条消息除了消息的Key和消息的值，还会增加两个字段:

分别是ProducerID和一个全局唯一的序列号，这个序列号由broker生成。

那么如下图所示，闪电表示消息发送后的ack确认失败，消息重传，Kafka会根据

对于第二点:broker的更新操作

需要实现把"消息读入->消息处理->结果写出"作为事务操作，也就是说这个操作需要满足ACID。

那Kafka是如何在保证系统性能的基础上实现事务的呢？

左下角表示事务日志，系统会存在一个事务锁，在某一个事务开始之前需要获得这个锁。

首先告诉系统开始一个事务，然后接着发送消息到brker相应的topic和partition，但是这些消息暂时是不能被消费方消费的，只有在所有操作成功完成之后再提交这个事务，这时候才标记这些消息是可以消费。

所有过程都会涉及到相应的log，只有在完成了这个事务之后，消费者才能消费这个事务所提交的消息。这么做保证了消息发送和业务逻辑要么全做，要么不做。不做意味着要支持事务的回滚，而事务日志是保存在相应的topic中的。

第三点：消费端如何保证

1.Kafka消费者只会读取那些被成功提交了的数据。所谓成功提交指的是消息的partition的leader和其所有的follower成功记录。这一方面也保证了消息的高可用。

2.消费端再消费数据的时候会提交偏移量到zk，然后broker根据zk的变化情况把相应数据标记为"已消费"。

正常情况下就不会出现重复消费的情况。那如果向zk提交偏移量之后服务宕机了，即broker里的数据没有标记为已消费，那么服务重启之后也会发生重复消费啊！

可以给消息分配一个全局ID，然后建立一个本地消息表，把提交zk偏移量和写入本地消息表放在一个事务进行，然后消费的时候再去查表看有没有消费过。

流处理系统架构

该部分是《数据密集型应用系统设计》关语流处理的部分

首先，批处理和流处理有何不同？

批处理的假设是：输入有界，如MapReduce的排序操作必须是先读取整个输入然后才能开始生成输出。

而流处理所处理的数据是无限的，批处理需要人为地把数据划分为固定时间段的数据块，如每天结束时候来处理一天的数据，或者在每个小时结束的时候处理这一个小时的数据。这种方法毫无疑问效率是很低的，那么流处理的思想就是为了减少这种延迟，每当有事件就可以开始处理。

一般来说"流"是指随着时间推移而持续可用的数据。所以可以把流当作一种数据管理机制一种无界，持续增量的处理方式。

接下来着重解决的问题是，如何通过网络来表示，存储和传输流？

流和数据库又有什么关系？

最后在处理流的过程中又有什么方法？

发送事件流

在批处理的世界里，作业的输入和输出是文件。

而对于流处理，当输入是文件(字节序列)的时候，第一个处理步骤通常是将其解析为记录序列，每一个记录可以理解为一个小的，独立的，不可变的事件对象。每一个事件对象会包含一个时间戳。

那怎么表示这个事件呢？

事件可以编码为文本字符串orJSONor某种二进制形式，那么事件就可以追加到文件，插入到关系表，或者将其写入文档数据库。

那么传统批处理是怎么通知消费者的呢？生产者把事件写入存储，然后每个消费者定期轮询数据存储来检查自上次运行以来出现的事件。但如果是流处理这么做的话，轮询的次数会变多，代价会变得很大。

消息系统

因此在流处理系统中，常见方法是使用消息系统，实现异步传输事件。

那不妨思考一下两个问题：

1.如果生产者发送的速度远远快于消费者所处理的速度：
可以采取以下三种策略：

1.系统丢弃消息

2.消息缓存在队列中
那如果内存无法满足无限增长的队列怎么办？能不能写入磁盘？写入磁盘又会不会影响消息传递系统的性能

3.激活背压，流量控制来阻止生产者发送更多消息
如TCP的窗口机制就是一种背压机制。

2 如果节点崩溃或暂时离线
所以要考虑数据的持久化会如何影响消息队列的性能？要么是进行多副本复制，或者是落盘，而这无疑会对消息队列的性能产生影响。

那生产者把消息写入消息队列和写入数据库有啥区别呢？

数据库会一直保留数据直到被明确要求删除,而大多数消息队列在消息成功传递时候就会自动删除消息。而批处理的一个关键特征是可以反复获取数据，不断重复尝试处理步骤。
如果消息量很大，而消费速度是很慢的，消息队列需要缓存很多消息的话，内存如果满足不了这个需求，就需要落盘。
查询数据库的时候，结果通常基于数据的时间点快照，如果另外一个客户端随后向数据库写入更改查询结果的内容，那么第一个客户端时不会察觉到数据已经过期的，除非重复查询或者轮询更改。而消息队列一旦发生数据变化便会告知客户端，实时性更强。

消息队列的确认和重新传递机制

消费者可能会随时崩溃。例如：队列向消费者传递消息，但是消费者从来不处理消息，或者在崩溃之前只对消息进行了部分处理。

为了确保消息不会丢失，消息代理需要使用确认机制：即客户端处理完消息后显式告诉代理，以便代理可以将其从队列中移除。

那么像Kafka，引入了偏移量机制。消费者在消费数据的时候，要提交偏移量到zk，然后Kafka再根据偏移量的变动把已消费的数据给删掉。

如果消息队列一直没有收到确认，那么就会认为消息未处理，因此消息将会传递给另外一个消费者。

那这样子会不会打乱另外一个消费者的消息处理顺序呢？那就不要使用负载均衡这种消息传递模式好了，改用扇出式，即每一条消息都传递给所有的消费者。（负载均衡式每一条消息只会传递给其中一个消费者）然后每个消费者维护一个单独的队列即可。

基于日志的消息存储

那既然基于文件系统和数据库的存储可以任意读取以前写入的数据，而基于消息系统只能接受订阅系统之后的数据，任何之前的数据都会消失，为何不把两者结合起来？

像Kafka就是一个经典的基于日志的消息存储系统例子：

日志是什么？
日志是磁盘上一个仅支持追加式修改记录的序列

生产者和消费者工作模式？
生产者通过把消息追加到日志的末尾来发送消息，而消费者通过依次读取日志来接收消息。

那如何突破单个磁盘能够提供的带宽吞吐上限？
采用分区机制，对日志进行分区，不同节点负责不同的分区。每个分区成为一个单独的日志，并且可以独立于其他分区的读取和写入。一个Topic包含了分布在不同节点上的分区。

消息的顺序是怎样的？
每个消息分配一个单调递增的序列号或者偏移量，那么分区只能追加，保证分区内部的消息是完全有序的，不同分区之间没有顺序保证。

消费者组的概念：
为了保证在一组消费者之间实现负载均衡，代理可以把整个分区分配给消费者组的节点，但是同一分区内的消息只会传递到同一个节点。那么这个节点会以单线程的方式顺序读取分区中的消息。

消费者偏移量的引入：
消息队列不需要跟踪每条消息的确认，只需要定期记录消费者的偏移量。那么消费者处理完业务之后，提交偏移量，消息队列就可以通过偏移量的变动来得知哪些消息已经被消费完了。

但有一个问题是如果消费者已经处理了业务，但在提交偏移量之前失败了，消息队列并没有标记这些数据是已经消费的，这时候消费者重启会发生重复消费的现象。
如何解决？

流处理

那么有了流之后，可以用它来做什么呢？

1.可以把事件中的数据写入数据库，缓存或者搜索索引或者类似的存储系统。

2.可以以某种方式推送给用户，如通过发送电子邮件警报或者推送通知

3.处理流以生成更多的流

CEP 复杂事件处理

与数据库持久存储不同的是，查询是暂时的，而数据是永久的。

而流系统的查询是长久存储的，来自输入流的事件会不断流过他们以匹配事件模式。

搜索同理，搜索条件是固定的，然后数据流过这个搜索条件。

流分析

例如测量某流数据的平均值，或者某种类型事件的速率。这些统计信息需要在固定的时间间隔内进行计算，这也是为什么流计算系统往往要引入窗口机制

流处理的时间

Flink其实有三种时间：

1.事件时间事件发生的时间
2.接入时间事件发送到服务器的时间
3.处理时间服务器收到事件的时间

那为什么要引入这些时间呢？这里着重讲一下事件时间和处理时间。因为流系统中网络很可能是不可靠的，按序生成的事件很有可能不是以生成的时间戳顺序接受处理。

所以引入事件时间，能够让系统得知流数据真正的到达时间的顺序。对于数据处理顺序有严格要求顺序的场景来说，混淆处理时间和事件时间将是致命的。

又举个例子：
如果以处理时间作为请求频率的测量，而不是事件时间，那么可能在系统重启这段过程中，挤压了许多请求，重启之后会出现一阵流量洪峰。但事实上请求频率一直是稳定的。

为什么要引入一个接入时间呢？因为用户控制的设备上的时钟通常是不可信的，可能会被意外或者故意设置成错误的时间。但是服务器收到事件的时间是我可以控制的，是可信的。那么通过处理时间-接入时间得到一个偏移量，就知道设备时钟和服务器时钟之间的偏移量（不考虑网路延迟），然后就可以根据偏移量正确得知事件真实发生的时间。

流和流join

两个输入流都由活动事件组成，采用join操作来搜索在特定时间窗口内发生的相关事件。

流和表join

一个输入流由活动事件组成，另外一个就是数据库变更日志，更新日志维护了数据库的本地最新副本。对于每个活动事件，join用来查询数据库并且输出一个包含更多事件的信息。

表和表join

两个输入流都是数据库的更新日志，结果就是两个表之间join的物化视图进行持续的更新

流处理的容错

如果MR作业中任务失败，可以简单在另一台机器上重新启动，并且丢弃失败任务的输出。最后任务输出到HDFS上的文件，而且输出仅在任务完成时候可见。

但是流是无限的，不能等到任务完成之后才得到输出，因为任务是永远不能完成的。

Flink的做法是定期生成状态的checkpoint并且将其写入持久化存储。如果流操作发生崩溃，可以从最近的检查点重新启动，并且丢弃上个检查点和崩溃之间生成的所有输出。

但是这都是在系统内部保证Exactly-Once,如何保证写入外部系统的时候也要实现端到端的Exactly-Once呢？

一种做法是Flink于外部系统交接的时候，采用两阶段原子提交方法，保证操作的事务性。
或者也可以利用消息队列的消息事务特性。
或者是第三方的软件和flink一样同样也维护checkpoint

消息中间件Kafka

Zookeeper

提供一个分布式的协调服务，基本单位为znode,把znode组织成一个树，提供：集群管理，命名服务，主节点选举，分布式锁等服务。

集群管理：监控节点的存货状态，运行请求。客户端对某个znode建立watcher监听，znode发生变化的时候，客户端会受到zookeeper的通知。
命名服务：通过组织成一颗树，可以定义全局唯一一个不重名的名字
分布式锁：提供两种锁实现，本质上是申请一个znode.独占锁和共享锁。
主节点选举：主节点挂掉后可以从备用节点开始新一轮选主，使用zookeeper可以协助完成这个过程。

kafka是什么？

高吞吐量的消息队列，可以理解为异步rpc，使用消息队列主要用来做业务的解耦和流量的削峰。

Topic

每一个消息都有其topic，Kafka通过topic对消息进行归类，kafka可以将Topic从物理上划分一个或者多个分区。每个分区物理上对应一个文件夹。

Partition

每一个分区都是一个顺序，不可变的消息队列，并且可以持续增加。分区中的消息都被分到一个序列号，称之为偏移量，在每个分区中偏移量都是唯一的。

消息存储在对应的分区上，可以为每条消息指定Key,这样消息被发送到Broker的时候，会根据分区算法把消息存储到对应分区。如果分区设置合理，那么就可以实现负载均衡。

Broker

Kafka Server，用来存储消息，消费者从Broker拉取消息

Producer

向Kafka发送消息，生产者会根据topic分发消息，生产者也会负责把消息关联到Topic的分区。最简单方法是从分区列表中轮流选择。也可以根据某种算法依照权重选择分区

Consumer

Consumer实例可以是独立的进程，负责订阅和消费信息，消费者用consumerGroup来标识自己。同一个消费组也可以并发消费多个分区信息。

分析一下kafka性能的瓶颈

cpu性能瓶颈
磁盘读写瓶颈
网络瓶颈

Kafka底层实现？为什么吞吐量那么高？

主要是通过把数据保存到磁盘上来实现强大的存储能力。我们常说的磁盘IO慢，其实是指随机读写IO很慢，但是如果是顺序IO的话
还是很快的。

而且采用了零拷贝技术：

通常kafka要发送数据给别的主机的时候，要经历以下步骤

1.磁盘把数据copy到OS 的页缓存
2.从缓存复制到kafka进程的缓存(从内核区copy到用户区)
3.用户进程再把数据写入到socket，数据流入内核区的socket buffer
4.再从socket buffer复制到网卡的copy

零拷贝技术直接跳过了第二第三步，数据可以直接从页缓存直接到网卡，那么从4次系统调用，2次上下文切换变回2次系统调用，1次上下文切换。

服务端优化

日志的存储
Kafka利用分段和追加日志的方法，把读写限制为磁盘的顺序IO而非随机IO
记录的批处理
顺序IO在大多数存储介质上很快，所以Kafka的性能瓶颈往往在网络而不是硬盘。所以客户端和服务端可以使用分批处理读写记录。
那么可以减少网络往返的次数，提高带宽效率。
批量压缩
启用压缩，同样也可以提高网络和磁盘IO的效率
便宜的消费者
Kafka不会在消息消费后删除消息，而是保留在硬盘一段时间(默认两天)，然后会独立跟踪每个消费者组的偏移量。结合Kafka的消息压缩，消息大小大大减少，只需要保留每一个消费者组对应的最后偏移量。意味着大量消费者可以并发从同一主题读取数据
未刷新的缓冲写操作
Kafka在确认写IO缓冲区的时就会返回ACK。而不会调用fsync这个系统调用，那么性能上有了很大的提高。所以本质上Kafka是一个磁盘支持的内存队列，只不过受到缓冲区和页面缓存大小的限制。

但是这样不能保证数据安全，但其实Kafka本身就不用保证写入来确保持久化，关键还是靠In-Sync Replica即同步副本机制。

换句话，无fsync的非阻塞IO方法和冗余的同步副本组合为Kafka提供了高吞吐，持久性和可用性。

客户端优化

零拷贝技术

1.初始read()方法导致上下文从用户态切换到内核态，文件被读取，内容被DMA复制到内核空间的页缓存区

2.把数据从内存到页缓存区复制到Kafka进程缓冲区

3.切换回内核态，把进程缓冲区的数据复制回内核空间的套接字缓冲区

4.套接字缓冲区复制回网卡

涉及到了四次上下文切换，在操作完成之前要复制四次数据（涉及到从用户态和内核态之间的复制）

什么是零拷贝呢?

无需经过套接字缓冲区，网卡会被赋予一个指向读缓冲区的指针，连同偏移量和长度，CPU是不涉及复制缓冲区的。

流处理的并行性

消费者组：消费者组里面的消费者实现负载均衡，每个消费者都是接收到不同的消息，整个消费者组来承担消费整个topic所有分区数据的概念。所以分区数目越多，能够消费的消费者数量也越多，间接提高了消费的数量。
而且这也是一种容错的消费机制，万一消费者组里面的某一个消费者挂了，可以由组内另外一个消费者来承担继续消费的任务。

Kafka使用一种负载均衡机制：把分区平均分配到各个消费者中。传统消息队列是利用副本机制(即N个消费者都是消费相同的消息)来提高可用性
而Kafka利用的是负载均衡来提高可用性持久性和吞吐量。

Kafka保证一个分区最多只能分配给消费者组中的一个消费者。（为什么用”最多“，当所有消费者都离线时，那就是0个消费者了。）当组中的第一个消费者订阅主题时，它将接收该主题上的所有分区。当第二个消费者订阅主题时，它将接收到大约一半的分区，从而减轻第一个消费者的负载。根据需要添加消费者（理想情况下，使用自动伸缩机制），这使你能够并行地处理事件流，前提是你已经对事件流进行了分区。可以增加消费者的数量来均衡入站记录的负载，消费者的数量最多可以增加到和分区数量一样多。（你可以增加更多的消费者，但每个分区最多只能有一个的活动消费者，剩下的消费者将处于闲置状态。）请注意，你可以提供一个线程池，根据消费者执行工作负载的不同，消费者可以是一个进程或一个线程

总结

1.日志追加方式记录数据，利用磁盘的顺序写
2.零拷贝IO避免缓冲区中的数据复制
3.分批消费
4.压缩算法
5.消费者组实现负载均衡,生产者按照key发往指定分区实现并行的流处理

调优思路：

1.提高消费的线程数目

2.提高分区数，提高吞吐率

3.配置文件调优:
增加网络IO线程数量和套接字数量
提高刷写磁盘的效率
增大linux内核中页缓存的大小

4.发送消息到broker之后，不需要等到broker确认，牺牲了数据的有序性和完整性

5.分批发送数据，增大每次发送数据的batch size，从而减少了发送的次数

如何保证Kafka消费的幂等性

写数据:
1.写数据到Kafka的分区是追加有序写入的，每一条数据对应一个偏移量offset
2.同时把offset的值保存到zk的节点上

读数据:
1.消费者从分区获取数据进行消费
2.消费完成后把对应消息的偏移量提交到zk
3.分区再根据zk来把消息标记为已消费

正常工作情况下，上述机制就可以保证Kafka消费的幂等性

异常情况:
1.如果消费者在消费完成后，提交offset的时候宕机
2.Partition无法从zk得知已经消费的消息对应的offset
3.因此消费者重启之后可能会重复消费这部分数据

解决方案:
给消息增加一个全局ID
全局ID可以在数据库或者一个本地消息表中建立唯一的索引
由数据库本身来保证全局ID的唯一性

Kafka如何支持消息事务

Kafka0.11版本后引入消息事务特性，这对流处理系统中不同中间件怎么保持Exactly-Once提供了一种很好的解决方案。

事务消息
确保本地执行事务与消息发送是原子性的：
1.先发送一条消息给消息中间件
2.执行本地事务
3.只有在执行本地事务成功之后才发送提交确认给消息中间件。

Kafka在0.11版本后提供事务支持
引入以下概念
1.事务协调者,每一个实现事务的生产端都会被分配到一个事务的协调者(Transaction Coordinator)。

2.引入一个Kafka 内部的 Topic作为事务Log: 类似于消费管理offset偏移量的topic，事务topic本身也是持久化的，日志消息记录事务的状态信息。

3.引入控制消息(Control Messages)：这些消息是客户端产生的并写入到主题的特殊消息，但对于使用者来说不可见。它们是用来让broker告知消费者之前拉取的消息是否被原子性提交。

4.引入TransactionId：不同生产实例使用同一个TransactionId表示是同一个事务，可以跨Session的数据幂等发送。当具有相同Transaction ID的新的Producer实例被创建且工作时，旧的且拥有相同Transaction ID的Producer将不再工作，避免事务僵死。

5.Producer ID：每个新的Producer在初始化的时候会被分配一个唯一的PID，这个PID对用户是不可见的。主要是为提供幂等性时引入的。

6.Sequence Numbler。（对于每个PID，该Producer发送数据的每个<Topic, Partition>都对应一个从0开始单调递增的Sequence Number。

7.每个生产者增加一个epoch：用于标识同一个事务Id在一次事务中的epoch，每次初始化事务时会递增，从而让服务端可以知道生产者请求是否旧的请求。

8.幂等性：保证发送单个分区的消息只会发送一次，不会出现重复消息。增加一个幂等性的开关enable.idempotence，可以独立与事务使用，即可以只开启幂等但不开启事务。

Kafka如何保证数据可靠性与一致性

思考角度:

Producer往Broker发送消息(消息确认机制)

1.ack=0 Best-Effort 只要生产者能够把信息发送出去就认为消息成功写入Kafka,这种模式吞吐率最高.但也会丢失一定的信息.

2.ack=1 在leader的副本收到消息(这里的收到消息指的是消息写入页缓存,而不是落盘,因为落盘要涉及到fsync系统调用)之后就会返回确认或者错误的响应.但是这种情况也有可能发生数据丢失,例如在消息复制到follower副本之前leader宕机了.

3.ack=all 可以保证所有同步副本都收到这个信息才返回确认.
如果和 min.insync.replicas 参数结合起来，就可以决定在返回确认前至少有多少个副本能够收到悄息，生产者会一直重试直到消息被成功提交。不过这也是最慢的做法，因为生产者在继续发送其他消息之前需要等待所有副本都收到当前的消息

Topic分区副本机制

Kafka可以保证单个分区内部的事件消费顺序有序的.分区可以在线(可用),也可以离线(不可用).分区副本有leader,有follower,所有的读写过程都经过Leader进行.

Kafka会为每个分区维护一个in-sync Replica副本集合,在这个集合里面都是follower,follower会异步从leader更新最新的数据.
所以Kafka是通过引入数据冗余来提供数据的可靠性.

Kafka 的分区多副本架构是 Kafka 可靠性保证的核心，把消息写入多个副本可以使 Kafka 在发生崩溃时仍能保证消息的持久性

Leader选举

每个分区的 leader 会维护一个 ISR 列表，ISR 列表里面就是 follower 副本的 Borker 编号，只有跟得上 Leader 的 follower 副本才能加入到 ISR 里面，这个是通过 replica.lag.time.max.ms 参数配置的，具体可以参见《一文了解 Kafka 的副本复制机制》。只有 ISR 里的成员才有被选为 leader 的可能。

所以当 Leader 挂掉了，而且 unclean.leader.election.enable=false 的情况下，Kafka 会从 ISR 列表中选择第一个 follower 作为新的 Leader，因为这个分区拥有最新的已经 committed 的消息。通过这个可以保证已经 committed 的消息的数据可靠性。

综上所述，为了保证数据的可靠性，我们最少需要配置一下几个参数：

producer 级别：acks=all（或者 request.required.acks=-1），同时发生模式为同步 producer.type=sync
topic 级别：设置 replication.factor>=3，并且 min.insync.replicas>=2；
broker 级别：关闭不完全的 Leader 选举，即 unclean.leader.election.enable=false；

Kafka怎么保证Consumer可以从新的Leader或者老的Leader都可以读取到一样的数据?

只有 High Water Mark 以上的消息才支持 Consumer 读取，而 High Water Mark 取决于 ISR 列表里面偏移量最小的分区，对应于上图的副本2，这个很类似于木桶原理.

这样做的原因是还没有被足够多副本复制的消息被认为是“不安全”的，如果 Leader 发生崩溃，另一个副本成为新 Leader，那么这些消息很可能丢失了。如果我们允许消费者读取这些消息，可能就会破坏一致性。试想，一个消费者从当前 Leader（副本0）读取并处理了 Message4，这个时候 Leader 挂掉了，选举了副本1为新的 Leader，这时候另一个消费者再去从新的 Leader 读取消息，发现这个消息其实并不存在，这就导致了数据不一致性问题.

流存储Pravega

Pravega应用场景

流处理对短时间内就能够连续生成的数据进行分析产生价值，而不需要等待批处理中累积和处理。核心优势是从摄取数据到计算数据的低延迟

举例子：车载系统的反馈分析，集群性能日志数据的分析告警，金融欺诈风控的精确定位，物联网煤气泄漏等。

在Pravega之前的流处理：

Lambda架构

特点：大数据平台分割成批处理层，流处理层，和应用服务层。

遵循读写分离，复杂性隔离的原则，整合了离线计算和实时计算。

Kappa架构

特点：把批处理，流处理简化为一致性的流处理。

Dataflow模型

Flink实现了真正的流批统一。而Spark 也推翻了之前微批处理的设计，使用Table和SQL进行流批统一，提出了Spark Structured Streaming。

Pravega的设计宗旨在于：

1.为流的实时存储解决方案，应用程序通过把数据持久化存储到Pravega。
2.Pravega的Stream可以提供无限制的数量并且持久化存储任意长的时间。而传统Kappa架构流处理应用的数据是通过消息队列来提供数据临时缓冲的，不提供持久化功能。
3.Pravega支持仅一次处理，可以在Kappa架构上实现链接应用需求。即可以把计算拆分为多个独立的微服务。
4.那么如果结合Pravega和Flink，整个流处理链路中写，读，存储过程都可以分开，是独立的。并且各个操作可以根据到达数据量来进行实时的动态扩展。所以Pravega的引入可以为Kappa架构形成了计算和存储的闭环。

流存储所要提供的功能:

下面把Kafka和Pravega两者进行比较

要把数据视作连续和无限的

Kafka如何模拟连续和无限的数据流？

通过采用添加到文件的末尾并且追踪其内容的方式(offset)来模拟连续和无限的数据流

缺点：
受限于本地文件系统文件描述符与磁盘容量制约。

Kafka怎么保证高可用？即数据的可靠性？

通过维护一个同步副本集合(in-sync)

Kafka中读写都是针对主partition进行的，然后维护一个同步副本集合，异步地从主副本实时同步数据。

这种做法无疑占用了更多的存储，并且其利用了消息头部的header记录元数据以构造数据结构，使得其不如字节序列那样通用。

Pravega 将从数据的角度支持的连续和无限的特点：

1.Pravega基本单位Stream是一个命名的，持久的，仅追加的，无限的字节队列。
2.提供尾读和追赶读

基于负载的自动弹性伸缩特性

Kafka通过分区这个概念来实现并行处理，类似的Hadoop也是利用了分区在HDFS和MapReduce实现了并行化的批处理。

缺点：分区会同时影响读客户端和写客户端

Pravega引入Segment概念来作为分区：

1.写客户端同时追加写不相交的数据子集

写入数据依靠路由键 (routing key) 写入不同的 segment 以保证隔离性。

让应用程序为写客户端分配键。

2.读客户端同时处理不相交的数据子集

读取的数据分区不依赖于写入分区

segment相比静态的分区，其数量会根据摄取的流量来自动连续更新。

连续处理数据以生成准确的结果(exactly-once)

具体来说

要考虑：

1.持久化

与Kafka消费数据之后就会从磁盘中删除该数据相比，而Pravega会把数据写入课持久化的分层存储来保证持久性。

2.有序性

Kafka能保证分区内部消费顺序是有序的，也就是说消费者组内消费是有序的。

什么是消费者组？

Consumer Group，同样是逻辑上的概念，是Kafka实现单播和广播两种消息模型的手段。同一个topic的数据，会广播给不同的group；同一个group中的worker，只有一个worker能拿到这个数据。换句话说，对于同一个topic，每个group都可以拿到同样的所有数据，但是数据进入group后只能被其中的一个worker消费。group内的worker可以使用多线程或多进程来实现，也可以将进程分散在多台机器上，worker的数量通常不超过partition的数量，且二者最好保持整数倍关系，因为Kafka在设计时假定了一个partition只能被一个worker消费（同一group内).

类似的，对于 Pravega 这样的通过路由键 (routing key) 来实现分区的系统而言，有序仅对具有相同键的数据有意义

3.一致性

就是指面对组件故障的时候，所有的读客户端都会看到给定的键相同的有序数据视图。当然仅仅靠组件内部实现一致性是不足够的，对于Pravega而言实现了强一致性。

4.怎么实现端到端的事务？

Pravega也是支持事务性的写入，使得在于Flink集成的时候，在Flink检查点之间建立事务，通过2PC来实现端到端的事务和仅一次处理。

Flink Connector

这部分也是我在Dell EMC 实习的记录，实习完了再弄吧
tbc

ref

kafka消息事务
https://mp.weixin.qq.com/s/3SMF1kYVutLFFcbCUc5fYQ
https://www.cnblogs.com/bethunebtj/p/9168274.html