kafka 在大数据领域应用及端到端一致性机制

    kafka在大数据架构应用很广，在大数据领域，特别是实时处理，由于数据量大，QPS高，在架构设计，都会采用kafka结合应用，用kafka暂存大量数据，然后业务端才会接入kafka开始处理业务逻辑，那kafka有哪些优点：

    1、高吞吐、低延迟：kafka每秒可以处理几十万消息，它延迟最低只有几毫秒。

     2、可扩展性，kafka集群支持热扩展，不同业务端，只要设定不同的groupId，即可快速接入kakfa来处理业务逻辑

    3、持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失。

     4、容错性高：允许集群节点故障（如果副本数为n，那允许n-1个节点故障）

    5、高并发：支持数千个客户同时读写。

在大数据领域的常见的架构设计如下：

     1、离线数仓架构图：

         以上业内常见的数仓架构，利用kafka收集各种服务的log与业务数据，通过kafka将数据消费到HDFS文件系统，然后通过HIVE/Tez等查询引擎做数据维度分层建模，最后提供分析决策支持，

        2、实时数据处理架构： 

   在实时数据处理框架中，kafka收集用户行为数据，而flink/spark也会开启相应的partition分区服务，实时/近实时接收日志数据，做用户行为偏好的统计与分析，输出特征数据或者对数据湖中的数据做相应的merge操作。

    对于以上架构，在本人的实际应用中，日志数据QPS达到了10万级别的QPS，双11大促峰值甚至达到了50万。所以kafka在大数据的庞大数据流处理这块，作用还是非常大。

kafka的机制学习

基于以上kafka的重大的应用价值，本人总结一些kafka的

offset存储位置，

如果是有zookeeper的情况，每一个partition的offset会更新到zookeeper的【consumers/{group}/offsets/{topic}/{partition}】目录下，可以通过以下命令查看：

get /kafka/consumers/martin-group_id/offsets/martiin-topic/0

在新版的kafka，kafka可以自己内部存储并管理offset，即可以不结合zookeeper使用，则offset的存储在kafka的内置的topic：__consumer_offsets，key:consumerGroupId+topic+分区号，value就是当前的offset值。而分区号kafka默认分配50个，可以通过offsets.topic.num.partitions设置

kafka offset管理不当会引发以下问题：

        重复消费：当数据已经被处理，然后自动提交offset时，消费者端程序出现故障崩溃或者有新的消费者加入导致再平衡，这时候提交offset失败，导致已经处理的数据的offset没记录，后续会出现重复消费

        丢失数据：如果offset设置了自动提交，如果业务处理时间过长，超过offset提交的规定时间，则offset已经提交了，但业务数据处理出现了异常崩溃，导致这些消息没有正常消费，由于offset已更新，后面处理不再重新这批数据，就会出现数据丢失的情况

flink+kafka如何实现端到端的一致性

     当分布式系统引入状态时，即kafka来的数据经历了flink多个算子处理后，仍能保持前后一致的offset位置，在某个算子出现故障后，后续恢复时，希望仍能恢复与前面的算子一致的offset状态。

     flink是有状态的流处理，每个算子任务都有自己的状态。

      对于流处理内部来说，所谓的状态一致性，就是计算结果在发生故障与没有故障时都是保证准确，一条数据不应该丢失，也不应该重复处理。

      状态一致性的级别如下：

       1、AT-MOST-ONCE（最多一次）

            当任务出现故障，最简单的事件什么事情都不做，就像UDP协议一样，既不恢复丢失的状态，也不重播丢失的数据，在kafka的状态管理的场景下，在处理数据前，offset已经提交了，导致在处理数据出现异常时，就没法再恢复之前offset，也就丢失了数据

       2、AT-LEAST-ONCE（最少一次）

           在大多数真实场景下，我们是不希望丢失数据，这种类型的保障称为AT-LEAST-ONCE(最少一次)在kafka的状态管理场景下，即在处理完数据后，才会提交提交数据，如果处理数据过程中出现异常崩溃，那就会重复原来offset位置进行消费处理。

       3、EXACTLY-ONCE(精确一次)

       恰好处理一次是最严格的的级别，也是最难实现的，不仅仅意味着没事件丢失，还意味着每一条数据，在内部状态仅仅更新一次。在实际场景中，需要结合数据确认状态及容器的幂等状态来确保。

如何保证flink与kafka的端到端的一致性

       在理论上，我们认为一致性的保证应该是由flink流处理器内部实现的。但在真实应用中，由于数据源与输出设备的特性不同，所以一致性保证还是需要数据源(kafka)与输出到持久化系统配合

      flink+kafka端到端的一致性，意味着结果的正确性贯穿了整个流应用的始末，每一个组件或者算子都可以保证自己的一致性，这意味端到端的一致性级别取决于所有组件中一致性最弱的组件

      可以看以下表格：

可以看出，任意方式，即不采用幂等或者事务性写入，则取决于source端的是否可重置，如果是不可重置的消息源如mysql，则在发生故障，由于没有类似于offset的记录数据位置，checkpoint也无法记录原来的数据位置 ，则无法再重播消息，就无法重新处理有问题的数据，即只能是at-most-once。

我们看flink整个流程的一致性情况：

        内部保证-----利用checkpoint机制，把状态存盘，发生故障时候可以恢复，保证内部的一致。

        source端---kafka consumer做为source，可以将偏移量保存下来，如果后续任务出现故障，恢复的时候可以由连接器重置偏移量，重新消费数据，保证一致性

        sink端-需要保证从故障恢复时，数据不会重复写入外部系统。在这sink端，有两种实现方式：幂等写入和事务性写入。事务性写入又分为事务提交及两阶段提交。

        如果sink端是redis，由于redis不管是新增数据还是修改修改，都是set操作，所以默认是幂等性写入。

        如果sink端是kafka producer，可以采用两阶段提交sink，但需要实现一个TwoPhaseCommitSinkFunction

offset在fink状态中是如何管理的

 在flink状态中，kafka提交offset需要区分是否开启了checkpoint。

如果checkpoint关闭，commit offset要依赖于kafka客户端是否配置了auto commit，即设置参数

//开启是否自动提交
kafkaProps.setProperty("enable.auto.commit","true")
//自动提交的时间间隔
kafkaProps.setProperty("auto.commit.interval.ms", "500")

这样kafka client就会按照500的时间间隔提交offset到kafka。

未完待续