【Flink】FLink 写入kafka 中关于 Exactly-Once 的一些思考

待我称王封你为后i 2022-09-07 09:59 225阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]

# 1.概述 #

首先看看文章：[【Flink】介绍Flink中状态一致性的保证][Flink_Flink]

根据文章内容化，我们知道kafka写写入是2阶段提交。2阶段提交看起来挺令人迷惑的，其实就是分2中情况嘛。

## 1.1 sink带事务 ##

带事务的sink端，一般都MySQL,Oracle，Kafka等。

比如kafka操作，我们一般事务代码如下

public static void main(String[] args) throws InterruptedException, ExecutionException { 
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.31.186:9092");
            props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            props.put(ProducerConfig.RETRIES_CONFIG, "3");
            props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");
            Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
    
            producer.initTransactions();
    
            try { 
                producer.beginTransaction();
                for (int i = 0; i < 5; i++) { 
                    Future<RecordMetadata> send = producer
                            .send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
                    System.out.println(send.get().offset());
                    TimeUnit.MILLISECONDS.sleep(1000L);
                }
                producer.commitTransaction();
            } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) { 
                // We can't recover from these exceptions, so our only option is to close the producer and exit.
                producer.close();
            } catch (KafkaException e) { 
                // For all other exceptions, just abort the transaction and try again.
                producer.abortTransaction();
            }
            producer.close();
        }

可以看到核心就是

producer.initTransactions();
    try { 
         producer.beginTransaction();
         ProducerRecord aa = new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i))；
         Future<RecordMetadata> send = producer.send(aa);
         producer.commitTransaction();
    
     } catch (KafkaException e) { 
         // For all other exceptions, just abort the transaction and try again.
         producer.abortTransaction();
     }
     producer.close();
    }

这段代码只不过是在2阶段提交的时候做了一些包装罢了，没有什么特别深奥的。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 1]  
在上面这一步就相当于做了

这些操作，其实已经将数据写入到目标的组件了，但是有没有真正写入topic这个需要看对方怎么实现的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 2]  
然后这一步当所有的都完成了，那么才提交事务，其实就是做了提交事务操作

producer.commitTransaction();

然后如果出问题了，那么也仅仅是取消事务

catch (KafkaException e) { 
         // For all other exceptions, just abort the transaction and try again.
         producer.abortTransaction();
     }

其他支持事务的操作大概一样的。

### 1.1.1 flink事务 ###

假设flink要支持事务，你该怎么搞？

#### 1.1.1.1 state ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 3]

假设是这种，flink消费一批数据后，没有立即发送数据，而且在sink端缓存了一下，然后在做checkpoint的时候 将本次的state全保存起来，放到hdfs中了，那么就意味着持久化了，然后将本批次发送出去，然后如果运气好，一次就发送成功，如下

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 4]  
事务成功后，就可以删除引用这个state的指针，释放掉，不重复发送了。

如果运气不好，那么就会重复发送，这样就会出现这种情况

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 5]  
可以看到这种情况下，数据重复了，如果sink到的组件没有去重功能数据就是重复了，但是如果对方做了数据去重功能，那么就是可以达到幂等性，就是相当于发送了一次。

这样的比如

1.  mysql的主键，第二次发送肯定会主键冲突
2.  ES的version，肯定会直接忽略第二次发送 参考：[【elasticsearch】 基于\_version进行乐观锁并发控制][elasticsearch_ _version] 完美知识又串联在一起了。
3.  kafka的数据压缩去重

#### 1.1.1.2 flink不处理 ####

上面可以看到如果数据攒一批然后发送可能造成sink端的数据重复，或者实时性不够，那么这种就是直接发送。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 6]  
这种情况下，flink中每次来了一个数据都发送给对方，然后对方不直接放到要写入的topic，只是放到一个临时的topic，然后到了一段时间，checkpoin进行的时候，就将本次的state保存起来，然后提交事务。

这里一看就有几个问题。

1.  假设提交事务的时候，挂了，导致没有提交成功，那么可以再次提交，让kafka将数据写入到真正的topic
2.  如果提交一半失败了，类似如下方案  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 7]  
    这种情况下，再次提交肯定会出现数据重复了，需要存储组件做幂等性去重存储。
3.  如果flink提交的时候，任务挂了呢？那么就需要从state中恢复，恢复的时候，看到有个引用的事务没有提交，那么需要再次提交。

## 1.2 sink端不带事务 ##

sink端不带事务的比如hdfs，文件之类的。

### 1.2.1 临时文件&临时topic ###

如果是Kafka上面说了一种临时的topic方案。

或者可以是临时文件。比如要写入到kafka可以写一个  
假设kafka保存的文件是

topic/index-0-100.txt

我们可以写一个文件是存储事务的如下

topic/index-100-200.tx

然后提交事务的时候，只需要将文件改个名字就好了，非常原子性。

### 1.1.2 预写日志 ###

这种也比较常见，就是存储端将数据先写入到一个临时文件，然后自己在提交事务的时候，将数据真正写入。

# 2. kafka事务 #

因为还没了解够kafka事务，然后根据 [【kafka】 kafka 2.3 版本 生产者和消费者事务 案例][kafka_ kafka 2.3 _ _] 找个案例结论。

*  默认的普通消费者设置的是read\_uncommitted 可以读取到未提交的事务数据。
 *  消费者设置read\_uncommitted 可以读取到未提交的事务数据。
 *  消费者设置read\_committed 只有在生产者提交事务的时候，才可以读取到数据，如果事务取消了，那么读取不到数据。

我猜测了另外一种实现。

kafka的事务数据是直接写入到真正topic的，但是也将事务信息写入到内置事务topic中了。

真正topic  分区   当前offset   事务结束offset
    topic_lcc  1     100         200
    
    事务topic   当前事务ID    事务元数据
    事务topic   1001        分区1 start_offset=100 endOffset=200

然后

*  当事务提交的时候，清除这个事务信息
 *  当事务还没提交的时候，如果设置的是read\_uncommitted，那么假设消费者读取到100了，然后下次读取的时候，直接读取100-200，这样就读取到没提交的事务了。
 *  如果事务回滚了，那么，如果设置的是read\_committed，那么假设消费者读取到100了，然后下次读取的时候，先看下事务是否提交，如果没有提交，那么就阻塞或者直接跳过100-200的数据，或者读取后缓存起来，这个处理应该比较麻烦，万一这个事务一直不提交怎么办，有很多问题。
 *  如果事务回滚了，那么，如果设置的是read\_committed，那么假设消费者读取到100了，然后下次读取的时候，先看下事务是否提交，如果回滚了，那么就阻塞或者直接跳过100-200的数据，这个最简单。

# 2. 小结 #

可以看到如果想要真正的 Exactly-Once 需要source-> sink->外部组件，三个一起努力完成。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70]: /images/20220829/88f82d7e5eb84248b8c3373133b9b98d.png
[Flink_Flink]: https://blog.csdn.net/qq_21383435/article/details/119862095
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 1]: /images/20220829/abec82f42d2b4404b2c8df3e8f17545b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 2]: /images/20220829/220c6953f413470b917f4b6ac52482aa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 3]: /images/20220829/c628d5d7e4714fcf9d666e34066b3144.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 4]: /images/20220829/83fcb633d93d4fe0a2184191e8361ce4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 5]: /images/20220829/d435e6a65f7c4dd0b507703b475585ce.png
[elasticsearch_ _version]: https://blog.csdn.net/qq_21383435/article/details/108901663
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 6]: /images/20220829/3e611021491e4ec4bef4df3128d15d45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzIxMzgzNDM1_size_16_color_FFFFFF_t_70 7]: /images/20220829/874626e846ba4ab4b20b9e30d3868277.png
[kafka_ kafka 2.3 _ _]: https://blog.csdn.net/qq_21383435/article/details/119879453?spm=1001.2014.3001.5501