上篇文章咱们实测了利用 Flink 的「EXACTLY_ONCE」和「AT_LEAST_ONCE」语义，以流的方式读取 HDFS 上的数据，然后将数据写入到 Elasticsearch(下称 ES)。

过程中，通过调整 checkpoint 时间，以及两种不同的语义方式，来对比不同参数组合下，数据入库的总量变化情况(B站直播的时候又演示了一遍)。

从最后的测试结果来看，仅仅只是设置 Flink 的 CheckpoinMode 而不采取其他任何措施的话，写入到 ES 的数据量都会比期望的要多。

那么这一次，咱们把数据源从之前的 HDFS，换成你可能更感兴趣的 Kafka，同时增加 Spark 作为对比，看测试效果会不会变得不一样。

0. 测试设计

还是把上次测试过的那批相同的数据，我用 Spark 给写入到 Kafka 里面，然后，分别用 Flink 和 Spark 以流的方式去消费，然后再次写入到 ES。

这一次，我们不再用「手动」的方式去 cancel 停掉进程，而是通过进程在处理数据时，自然发生「处理错误」导致进程退出。

然后，对错误进行捕捉处理，再次对程序通过「checkpoint」恢复，看最终写入到 ES 的数据量跟预期的是否一致。

其中，经过调试之后的 Spark 代码如下：

package com.anryg.bigdata.streaming.es

import java.util.concurrent.TimeUnit

import com.alibaba.fastjson.JSON
import com.anryg.bigdata.clickhouse.CKSink
import com.anryg.bigdata.{IpSearch, RedisClientUtilsV2}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.{OutputMode, Trigger}

/**
  * @DESC: 通过spark structured streaming消费kafka写 ES，测试程序失败后，通过checkpoint恢复，数据重复的情况
  * @Auther: Anryg
  * @Date: 2024/4/16 10:18
  */
object SparkFromKafka2ES {

    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setAppName("SparkFromKafka2ES").setMaster("local[*]")
        val spark = SparkSession.builder().config(conf).getOrCreate()

        val rawDF = spark.readStream //获取数据源
                .format("kafka") //确定数据源的来源格式
                .option("kafka.bootstrap.servers", "192.168.211.107:6667") //指定kafka集群的地址，理论上写一个broker就可以了
                .option("subscribe","test")  //指定topic
                .option("failOnDataLoss",false)  //如果读取数据源时，发现数据突然缺失，比如被删，则是否马上抛出异常
                .option("fetchOffset.numRetries",3)  //获取消息的偏移量时，最多进行的重试次数
                .option("startingTimestamp","1713182435000") //通过时间戳指定消费位置
                .load()

        import spark.implicits._

        val ds = rawDF.selectExpr("CAST(value AS STRING)")  //将kafka中的数据的value转为为string，原始为binary类型
                .map(row => {
                    val line = row.getAs[String]("value") //获取row对象中的field，只有一个field
                    val array = line.split(",")  //指定分隔符进行字段切分

                    var rcode = 99
                    var query_type = 99
                    var client_ip = ""
                    var domain = ""
                    var time = ""
                    var target_ip = ""
                    var authority_record = ""
                    var add_msg = ""
                    var dns_ip = ""
                    try {
                        if (array(4).length == 1) rcode = array(4).toInt
                        if (!array(5).equals("")) query_type = array(5).toInt //制造可能的数据异常
                        client_ip = array(0)
                        domain = array(1)
                        time = array(2)
                        target_ip = array(3)
                        authority_record = array(6)
                        add_msg = array(7)
                        dns_ip = array(8)
                    } catch {
                        case e:Exception => {
                            println(line,e) //捕获异常数据异常情况
                            if (args(0).equals("0")) System.exit(-1) //控制遇到数据异常时，是否退出进程
                        }
                    }
                    (client_ip, domain, time, target_ip, rcode, query_type, authority_record, add_msg, dns_ip)
                }).toDF("client_ip", "domain","time","target_ip","rcode","query_type","authority_record","add_msg","dns_ip") //给裸数据添加字段名

        val query = ds.writeStream
                .outputMode(OutputMode.Append())  //指定数据的写入方式
                .format("org.elasticsearch.spark.sql")  //指定外部输出为ES
                .option("es.nodes","192.168.221.173")
                .option("es.port","9201")          
                .option("checkpointLocation","hdfs://192.168.211.106:8020/tmp/spark_checkpoint/SparkFromKafka2ES") /**用来保存offset，用该目录来绑定对应的offset，如果该目录发生改变则程序运行的id会发生变化，类比group.id的变化*/
                .start(args(1)) //指定目标索引

        query.awaitTermination()
    }
}

Flink 代码如下：

package com.anryg.es

import java.util
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.connector.sink2.SinkWriter
import org.apache.flink.connector.elasticsearch.sink.{Elasticsearch7SinkBuilder, ElasticsearchSink, RequestIndexer}
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend
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 DataStream API读取kafka数据，写入到ES，测试 checkpointmode 不同语义下，数据恢复的情况
  * @Auther: Anryg
  * @Date: 2024/4/16 17:05
  */
object FlinkDSFromKafka2ES {

    /**
      * @DESC: 设置checkpoint的mode
      * */
    private def setCheckpointMode(env:StreamExecutionEnvironment, mode:String): Unit ={
        if (mode.equals("1")) env.enableCheckpointing(10000, CheckpointingMode.EXACTLY_ONCE)
        else env.enableCheckpointing(10000, CheckpointingMode.AT_LEAST_ONCE)
    }

    def main(args: Array[String]): Unit = {
        //获取流任务的环境变量
        val env = StreamExecutionEnvironment.getExecutionEnvironment

        /**通过参数控制用哪种语义*/
        setCheckpointMode(env, args(0)) //设置checkpointmode

        env.setStateBackend(new EmbeddedRocksDBStateBackend(true)) //新的设置state backend的方式
        env.getCheckpointConfig.setCheckpointStorage("hdfs://192.168.211.106:8020/tmp/flink_checkpoint/FlinkDSFromKafka2ES") //设置checkpoint的hdfs目录
        env.getCheckpointConfig.setExternalizedCheckpointCleanup(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) //设置checkpoint记录的保留策略

        import org.apache.flink.streaming.api.scala._  //引入隐式转换函数,需要将从kafka读出来的数据做隐式转换

        val kafkaSourceBuilder = KafkaSource.builder()
            .setBootstrapServers("192.168.211.107:6667")
            .setTopics("test")
            .setGroupId("FlinkDSFromKafka2ES")
            .setStartingOffsets(OffsetsInitializer.timestamp(1713182435000L)) //指定消费的时间戳
            .setValueOnlyDeserializer(new SimpleStringSchema())

        import org.apache.flink.streaming.api.scala._  //引入隐式转换函数,需要将从kafka读出来的数据做隐式转换

        /**读取数据源生成DataStream对象*/
        val kafkaDS = env.fromSource[String](kafkaSourceBuilder.build(), WatermarkStrategy.noWatermarks(),"kafka-data")

        val builder = new Elasticsearch7SinkBuilder()
            .setBulkFlushMaxActions(args(1).toInt) //批写入大小
            .setHosts(new HttpHost("192.168.221.173",9201,"http"))
            .setEmitter((element:String, context:SinkWriter.Context, indexer: RequestIndexer) =>{

                val array = element.split(",")

                var rcode = 99
                var query_type = 99
                var client_ip = ""
                var domain = ""
                var time = ""
                var target_ip = ""
                var authority_record = ""
                var add_msg = ""
                var dns_ip = ""
                try {
                    if (array(4).length == 1) rcode = array(4).toInt
                    if (!array(5).equals("")) query_type = array(5).toInt  //制造可能的数据异常
                    client_ip = array(0)
                    domain = array(1)
                    time = array(2)
                    target_ip = array(3)
                    authority_record = array(6)
                    add_msg = array(7)
                    dns_ip = array(8)
                } catch {
                    case e:Exception => {
                        println( element,e) //捕获异常数据异常情况
                        if (args(2).equals("0")) System.exit(-1) //控制遇到数据异常时，是否退出进程
                    }
                }

                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.toString)
                map.put("query_type", query_type.toString)
                map.put("authority_record", authority_record)
                map.put("add_msg", add_msg)
                map.put("dns_ip", dns_ip)

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

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

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

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