来源：安瑞哥是码农

还记得第一次用 Flink 的时候，就被它设置 CheckpointingMode 时，可以选择用 AT_LEAST_ONCE 或者 EXACTLY_ONCE 恢复数据状态的模式给「惊」住了。

当然，这个惊，不是感叹 Flink 有多么厉害，而是感慨，保证「精确一次性」这么大言不惭的牛逼，是怎么敢吹出来的？

做过大数据开发的都应该知道，计算引擎在一个项目中，它从来都是处于一个「承上启下」的地位，上承数据源，下接数据库，连在一起，就是3个异构的系统在一起配合着干活。

对于任何一次数据计算，想要到达所谓的「精确一次性」，那就要保证这里面3个环节的「幂等」或者说「事务」，而我们最常见的，可以保证事务的系统，是单节点的数据库，比如 MySQL、Oracle这些。

它们之所以能保证这个「精确一次性」，原因在于人家有非常复杂、且完备的日志管理体系，最关键的是，人家只敢保证「在一个系统内部」。

而你 Flink，却要确保3个异构系统的精确一次性，凭啥？

虽然对于 Flink 的这个 EXACTLY_ONCE ，源码解释了它的含义，

之前我在一期「如何确保精确一次性消费」的视频里，也用工程化的思路，拿具体案例讲过它的解决方案，说想要保证这一点，一定不是简单配置某个参数，或者开启某个开关就能做到的，它是一个「系统化」工程。

但总架不住一些 Flink 的崇拜粉坚定地认为：这能有多难的？用 Flink 设置 EXACTLY_ONCE 模式不就解决了吗？还费这些劲？

说这话的人啊，只能说明一点：没有经历过真实环境的毒打，自己掌握的那点知识呢，也就仅仅停留在网络上口口相传的烂大街八股文里，但凡有点脑子跟怀疑精神，就不会这么武断。

那作为一个喜欢给那些「高大上」技术祛魅，热衷扯掉它们身上「遮羞布」的人，今天咱就来实地演示一下，这个 Flink 的 EXACTLY_ONCE 模式，到底有没有一些人认为的那么神？

0. 祛魅准备

因为这个 CheckpointingMode 的 EXACTLY_ONCE 模式是专门针对「流」场景的，所以，咱们也要找一个流模式下的测试方式。

从以往我测试过的 Flink 诸多场景来看，它既可以用流的方式消费「消息队列」数据，也可以消费「文件系统」数据。

那么今天，我们就设计用 Flink 流的方式读取 HDFS 的某个文件，

然后将数据经过简单的处理后，写入到 Elasticsearch (下称ES)。

测试过程大致如下：

第1步：Flink 以流的方式读取 HDFS 的全部数据，然后写入到 _id 自动生成的 ES 索引中，看整个数据写完，总数据量有多少条，跟原始数据量是否一致；

第2步：也是用 Flink 以流的方式读取 HDFS 的全部数据，把 CheckpointingMode 设置成 EXACTLY_ONCE 模式，写入到另一个同样 _id 自动生成的 ES 索引中，但是，在数据写入过程中，故意停掉进程，然后通过上一次的 checkpoint 再次恢复，看最终写入的数据量是否符合预期。

第3步：其他参数不变，将 EXACTLY_ONCE 换成 AT_LEAST_ONCE 再测试，看结果有哪些变化；

第4步：其他参数不变，只修改 checkpoint 的时间间隔，再看写入到 ES 的数据量，有多大变化。

测试样例代码如下：

package com.anryg.hdfs

import java.time.Duration
import java.util

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.connector.sink2.SinkWriter
import org.apache.flink.connector.elasticsearch.sink.{Elasticsearch7SinkBuilder, ElasticsearchSink, RequestIndexer}
import org.apache.flink.connector.file.src.FileSource
import org.apache.flink.connector.file.src.reader.TextLineInputFormat
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests


/**
  * @DESC: 对 Flink 的精确一次性消费进行测试
  * @Auther: Anryg
  * @Date: 2024/4/11 10:46
  */
object FlinkExactlyOnceTest {

    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment

        env.enableCheckpointing(args(0).toLong) //设置Checkpoint时间间隔

        env.setStateBackend(new EmbeddedRocksDBStateBackend(true)) //新的设置state backend的方式
        env.getCheckpointConfig.setCheckpointStorage("hdfs://192.168.211.106:8020/tmp/flink_checkpoint/FlinkExactlyOnceTest")
        env.getCheckpointConfig.setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) //设置checkpoint记录的保留策略
        env.getCheckpointConfig.setAlignedCheckpointTimeout(Duration.ofMinutes(1L))
        //控制用哪种Checkpoint模式
        if (args(1).equals("1")) env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
        else env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE)

        val fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat,new Path("hdfs://192.168.211.106:8020/tmp/from_doris/b5bb9f7420d84255-900a15f5988d44a3_0.csv"))
                                    .monitorContinuously(Duration.ofSeconds(3))
                                    .build()

        import org.apache.flink.streaming.api.scala._  //引入隐私转换函数

        val fileDS = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(),"file-from-hdfs-2ES")

        val builder = new Elasticsearch7SinkBuilder()
                .setHosts(new HttpHost("192.168.221.173",9201,"http"))
                .setEmitter((element:String, context:SinkWriter.Context, indexer: RequestIndexer) =>{

                    val array = element.split(",")

                    var client_ip = ""; var domain = ""; var time = ""; var target_ip = ""; var rcode = "99"; var query_type = "99"; var authority_record = ""; var add_msg = ""; var dns_ip = ""
                    try {
                        if (array(0) != null) client_ip = array(0)
                        if (array(1) != null) domain = array(1)
                        if (array(2) != null) time = array(2)
                        if (array(3) != null) target_ip = array(3)
                        if (array(4) != null) rcode = array(4)
                        if (array(5) != null) query_type = array(5)
                        if (array(6) != null) authority_record = array(6)
                        if (array(7) != null) add_msg = array(7)
                        if (array(8) != null) dns_ip = array(8)
                    } catch {
                        case e: Exception => println(element, e)
                    }

                    val map = new util.HashMap[String,String]() /**必须是Java的map类型*/
                    map.put("client_ip", client_ip)
                    map.put("domain", domain)
                    map.put("time", time)
                    map.put("target_ip", target_ip)
                    map.put("rcode", rcode)
                    map.put("query_type", query_type)
                    map.put("authority_record", authority_record)
                    map.put("add_msg", add_msg)
                    map.put("dns_ip", dns_ip)

                    val indexRequest = Requests.indexRequest()
                        .index(args(2)) /**目标索引*/
                        .source(map)

                    indexer.add(indexRequest)
                }
                ).build()

        fileDS.sinkTo(builder.asInstanceOf[ElasticsearchSink[String]])/**这里需要进行强制转换*/

        env.execute() //启动任务,action按钮
    }
}