Flink状态 和 Exactly Once_execution.checkpointing.interval:3min 配置

状态和普通变量的区别：
普通变量数据保存在内存中，任务执行失败会丢失
flink的状态中的数据会被checkpoint持久化到hdfs中，如果任务失败还能恢复到之前的计算结果
flink的checlpoint默认是关闭的，开启后在本地无法运行了，只能提交服务器了
    // 每 1000ms 开始一次 checkpoint
    env.enableCheckpointing(1000)
    // 高级选项：
    // 设置模式为精确一次 (这是默认值)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 确认 checkpoints 之间的时间会进行 500 ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    // Checkpoint 必须在一分钟内完成，否则就会被抛弃
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    // 允许两个连续的 checkpoint 错误
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)
    // 同一时间只允许一个 checkpoint 进行
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
    //RETAIN_ON_CANCELLATION: 当任务取消时保留checkpoint
    env.getCheckpointConfig.setExternalizedCheckpointCleanup(
      ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    // 需要设置flink checkpoint保存状态的位置
    env.setStateBackend(new HashMapStateBackend())
    //将状态保存到hdfs中  env.getCheckpointConfig.setCheckpointStorage("hdfs://master:9000/flink/checkpoint")

sum()算子，底层使用了flink的状态保存之前的计算结果，flink的状态会被checkpoint持久化到hdfs中，任务被取消或者执行失败可以恢复之前的计算结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

1. state

1. flink用于保存之前计算结果的机制
2. flink会为每一个key保存一个状态
3. 常用的sum（需要保存之前的计算结果）  window（需要保存一段时间内的数据）内部都是有状态的
4. flink也提供了几种常用的状态类
   1. valueState: 单值状态，为每一个key保存一个值，可以是任何类型，必须可以序列化
   2. mapState: kv格式的状态，为每一个key保存一个kv格式的状态
   3. listState: 集合状态，为每一个key保存一个集合状态，集合中可以保存多个元素
   4. reducingState/AggregatingState:聚合状态，为每一个key保存一个值，再定义状态时需要一个聚合函数
5.flink的状态和普通变量的区别：
 普通变量是保存再flink的内存中的，如果flink任务执行失败，变量的数据会丢失
 flink的状态是一个特殊的变量，状态中的数据会被checkpoint持久化到hdfs中, 如果任务执行失败，重启任务，可以恢复状态
6.状态后端，用于保存状态的位置
1. HashMapStateBackend： 
   -1. 将flink的状态先保存TaskManager的内存中，在触发checkpoint的时候将taskmanager中的状态再持久化到hdfs中
   -2. 可以直接使用
      ```java
      env.setStateBackend(new HashMapStateBackend())
      ```
2. EmbeddedRocksDBStateBackend：
  - 1. RocksDS是一个本地的轻量级的数据库，数据在磁盘上
  - 2. 再启动lfink任务的时候会在每一个taskManager所在的节点启动一个rocksDB进程
  - 3. flink的状态会先保存在rocksDb数据库中，当触发checkpoint的时候将数据库中的状态持久化到hdfs中
   -4. 可以支持增量快照
   -5. 使用rocksDb状态后端需要带入依赖
      ```xml
      <dependency>
          <groupId>org.apache.flink</groupId>
          <artifactId>flink-statebackend-rocksdb</artifactId>
          <version>1.15.0</version>
      </dependency>
      ```
   -6. 使用方式
      ```java
      env.setStateBackend(new EmbeddedRocksDBStateBackend(true))
      ```

      
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

valueState状态：存储之前的计算结果
多种状态

open()：在map之前执行，每一个task中只执行一次
flink的状态需要在open中定义

实时计算平均年龄：
object Demo14AvgAge {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

   //实时计算每一个班级的平均年龄
    val studentDS: DataStream[String] = env.socketTextStream("master", 8888)
    //取出班级和学生的年龄
    val clazzAndAge: DataStream[(String, Int)] = studentDS.map(line => {
      val split: Array[String] = line.split(",")
      val clazz: String = split(4)
      val age: Int = split(2).toInt
      (clazz, age)
    })
    //按照班级分组
    val keyByDS: KeyedStream[(String, Int), String] = clazzAndAge.keyBy(_._1)
    //计算平均年龄
    //flink的状态可以再任务算子中使用，map,faltmap,filter process都可以
    val avgAgeDS: DataStream[(String, Double)] = keyByDS
      .process(new KeyedProcessFunction[String, (String, Int), (String, Double)] {
        override def open(parameters: Configuration): Unit = {        
           /定义两个状态来保存总人数和总的年龄         
          val context: RuntimeContext = getRuntimeContext
          //总人数状态的描述对象
          val sumNumDesc = new ValueStateDescriptor[Int]("sumNum", classOf[Int])
          //总的年龄的描述对象
          val sumAgeDesc = new ValueStateDescriptor[Int]("sumAge", classOf[Int])
          sumNumState = context.getState(sumNumDesc)
          sumAgeState = context.getState(sumAgeDesc)
        }
        //保存总的人数的状态
        var sumNumState: ValueState[Int] = _
        //保存总的年龄的状态
        var sumAgeState: ValueState[Int] = _

        override def processElement(kv: (String, Int),
                                    ctx: KeyedProcessFunction[String, (String, Int), (String, Double)]#Context,
                                    out: Collector[(String, Double)]): Unit = {

          val clazz: String = kv._1
          val age: Int = kv._2

          //获取之前的总的人数和总的年龄
          var sumNum: Int = sumNumState.value()
          //人数累加
          sumNum += 1
          //更新状态
          sumNumState.update(sumNum)　
          var sumAge: Int = sumAgeState.value()
          //累加
          sumAge += age
          //更新状态
          sumAgeState.update(sumAge)
          //计算平均年龄
          val avgAge: Double = sumAge.toDouble / sumNum
          //将数据发送到下游
          out.collect((clazz, avgAge))
        }
      })　
    avgAgeDS.print()
    env.execute()
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

2. checkpoint

checkpoint是flink用于持久化flink状态的机制
flink会定时将flink计算的状态持久化到hdfs中
开启checkpint的方法： 代码中或者源码中

2.1 代码单独开启优先级最高

  // 每20000ms 开始一次 checkpoint
env.enableCheckpointing(20000)
// 高级选项：
// 设置模式为精确一次 (这是默认值)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
// 确认 checkpoints 之间的时间会进行 500 ms
env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
// Checkpoint 必须在一分钟内完成，否则就会被抛弃
env.getCheckpointConfig.setCheckpointTimeout(60000)
// 允许两个连续的 checkpoint 错误
env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)
// 同一时间只允许一个 checkpoint 进行
env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
// 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
//RETAIN_ON_CANCELLATION: 当任务取消时保留checkpoint
env.getCheckpointConfig.setExternalizedCheckpointCleanup(
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
//指定状态后端
//EmbeddedRocksDBStateBackend eocksDb状态后端或者new HashMapStateBackend()
env.setStateBackend(new EmbeddedRocksDBStateBackend(true))
//将状态保存到hdfs中，在触发checkpoint的时候将状态持久化到hdfs中
env.getCheckpointConfig.setCheckpointStorage("hdfs://master:9000/flink/checkpoint")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

2.2 在源码中修改配置开启（flink新版可以操作）

vim  flink-conf.yaml

execution.checkpointing.interval: 3min
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.min-pause: 0
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.timeout: 10min
execution.checkpointing.tolerable-failed-checkpoints: 0
state.backend: rocksdb
state.checkpoints.dir: hdfs://master:9000/flink/checkpoint
1
2
3
4
5
6
7
8
9
10
11

2.3 从checkpoint中恢复任务

1.可以在网页中指定checkpint的路径恢复,初次提交任务不需要指定路径，恢复任务时需要指定hdfs中保存状态的路径，路径需要带上前缀hdfs://master:9000
样式：hdfs://master:9000/flink/checkpoint/11edbec21742ceddebbb90f3e49f24b4/chk-35
2.在命令行中重新提交任务，指定恢复任务的位置：   -s  指定恢复任务的路径
flink run -t yarn-session -Dyarn.application.id=application_1658546198162_0005  -c com.shujia.flink.core.Demo15RocksDB -s hdfs://master:9000/flink/checkpoint/11edbec21742ceddebbb90f3e49f24b4/chk-35 flink-1.0.jar
1
2
3
4

3. EXACTLY ONCE（数据处理的唯一一次）

3.1 数据消费端（存任务失败时传来的数据，存状态）

注释：从Kafka中消费数据，数据来一条，计算一次
flink会将kafka消费偏移量和计算处理的状态存储在hdfs中，所以当任务失败时，即使map端读取过一次数据但是没有被下游消费，下次还是从状态中保存的位置开始处理

EXACTLY ONCE：指任务发生异常的情况下保证每一条数据只被处理一次，并不是指只被读取一次，保证最终结果不多不少，
1.最多一次：会丢失数据
至少一次：会有重复数据
2.实时计算中才可能会发生，离线计算不存在这种问题（失败了大不了重新计算），而流式处理不行，之前计算过的结果必须要保存好，而且不能重新开始进行计算，否则结果会出错
3.任务在其重启的时候，数据会被重复读取但是不会被重复进行计算，最终结果保证不错

3.2 数据生产端

Kafka如保证书数据不重复
Kafka0.11之后，生产者发送数据时幂等的，在任何导致任生产端重试的情况下，相同的消息被发送了多次也能保证只被写入kafka中一次，保证了数据不会重复
kafka如何保证数据不丢失：
1.副本存储在不同的节点中（多分区多副本），避免单节点故
障，副本间进行数据同步（问题：数据已经发送到第一个分区，在数据同步完成之前，节点挂了，数据就丢失了，怎么办？）
2.ACK机制（第二次提到了）：解决2的问题
acks=0时:生产者只负责生产数据，不管Kafka是否保存数据成功，生产效率高但是会丢失数据
acks=1时(默认):生产者等待第一个副本的数据保存成功再返回数据发送成功的状态，如果此时第一个副本数据所在的节点挂了，会导致数据丢失
acks=-1时:发送一批次数据，生产者需要等所有副本数据存储完成后才返回成功，不会丢失数据，但是效率低

3.事务支持：事务提交之前都可以进行回滚

3.3 通过flink读取kafka的数据后向kafka中写数据（Kafka章节中有写）

1.读写kafka数据都是实时的，数据无法退回来，但是数据时不会丢的
2.生产数据到kafka中时，默认的发送模式是至少一次，当任务执行失败时会有一部分数据发送多次，因此kafka-sink中会有重复数据，那么如何才能不重复呢？
3.kafka0.11之后支持：两步提交的SinkFunction，
4.hadoop不支持事务，MySQL，hbase支持事务

-----------------------------------------------------------------------------------
object Demo17ExactlyOnce {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
...2.1中的一堆开启checkpoint的代码...
val source: KafkaSource[String] = KafkaSource.builder[String]
      .setBootstrapServers("master:9092,node1:9092,node2:9092")
      .setTopics("source")
      .setGroupId("Demo16ExactlyOnce")
      .setStartingOffsets(OffsetsInitializer.earliest) 
 //只在第一次启动的时候生效，因为开启了checkpoint,任务重启之后会按照checkpoint中保证的偏移量消费数据，下一恢复数据有指定的hdfs路径
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build

    val kafkaSource: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
    //过滤空数据
    val filterDS: DataStream[String] = kafkaSource.filter(_.nonEmpty)

    val properties = new Properties()      //开启一个事务
    //设置事务的超时时间，要比15分钟小
    properties.setProperty("transaction.timeout.ms", 10 * 60 * 1000 + "")
    
    val kafkaSink: KafkaSink[String] = KafkaSink
      .builder[String]()
      .setBootstrapServers("master:9092,node1:9092,node2:9092") //broker地址
      .setKafkaProducerConfig(properties) //设置额外的参数
      .setRecordSerializer(
        KafkaRecordSerializationSchema
          .builder[String]()
          .setTopic("sink") //topic
          .setValueSerializationSchema(new SimpleStringSchema())
          .build())
      .setDeliverGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
      .build()
      
    filterDS.sinkTo(kafkaSink)    //将DS装到设置好的sink中
    env.execute()
  }
}
-----------------------------------------------------------------------------------
通过命令消费sink数据    Kafka消费数据控制台消费者模式：读已提交模式
--isolation-level read_committed : 只读已提交的数据
kafka-console-consumer.sh --bootstrap-server master:9092,node1:9092,node2:9092 --isolation-level read_committed --from-beginning --topic sink     
注释：这样的话就会在两个checkPoint之间开启一个事务，两个checkpoint同时读到的时候事务才算完成，否则任务失败回滚到之前的状态，即可保证不重复往Kafka-sink中写数据，相当于微批处理，数据有一定的延迟（开启了checkpoint,延迟和不重复只能选一个）

DeliveryGuarantee.EXACTLY_ONCE：   设置则为唯一一次
DeliveryGuarantee.AT_LEAST_ONCE：  默认是至少一次
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

3.4 消费kafka中的数据

object Demo16ExactlyOnce {
  def main(args: Array[String]): Unit = {
      //使用flink从kafka中读取数据，怎么保证数据处理的唯一一次
      val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    
      //开启checkpoint   
    // 每 1000ms 开始一次 checkpoint
    env.enableCheckpointing(20000)
    // 高级选项：
    // 设置模式为精确一次 (这是默认值)    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    // 确认 checkpoints 之间的时间会进行 500 ms
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500)
    // Checkpoint 必须在一分钟内完成，否则就会被抛弃
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    // 允许两个连续的 checkpoint 错误
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(2)
    // 同一时间只允许一个 checkpoint 进行
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
    //RETAIN_ON_CANCELLATION: 当任务取消时保留checkpoint
    env.getCheckpointConfig.setExternalizedCheckpointCleanup(
      ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)    
     // 需要设置flink checkpoint保存状态的位置        
    env.setStateBackend(new HashMapStateBackend())
    //将状态保存到hdfs中    env.getCheckpointConfig.setCheckpointStorage("hdfs://master:9000/flink/checkpoint")  

     // 消费kafka中的数据  
    val source: KafkaSource[String] = KafkaSource.builder[String]
      .setBootstrapServers("master:9092,node1:9092,node2:9092")
      .setTopics("words")
      .setGroupId("Demo16ExactlyOnce")
      .setStartingOffsets(OffsetsInitializer.earliest) //只在第一次启动的时候生效，如果开启了checkpoint,任务重启之后会按照checkpoint中保证的偏移量消费数据
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build

  //消费后处理数据
    val kafkaSource: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
    val wordsDS: DataStream[String] = kafkaSource.flatMap(_.split(","))
    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)
    val countDS: DataStream[(String, Int)] = keyByDS.sum(1)
    countDS.print()
    env.execute("Demo16ExactlyOnce")
注释：保存上次的消费偏移量就能保证数据只消费一次
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

3.5 数据写到kafka

object Demo6KafkaSInk {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val studentDS: DataStream[String] = env.readTextFile("data/students.json")
val sink: KafkaSink[String] = KafkaSink
      .builder[String]()
      .setBootstrapServers("master:9092,node1:9092,node2:9092") //broker地址
      .setRecordSerializer(
        KafkaRecordSerializationSchema
          .builder[String]()
          .setTopic("students_json") //topic
          .setValueSerializationSchema(new SimpleStringSchema())
          .build())
      .build()
    //使用kafka sink
    studentDS.sinkTo(sink)    
    env.execute()
  }
}

DeliveryGuarantee.EXACTLY_ONCE：   设置则为唯一一次
DeliveryGuarantee.AT_LEAST_ONCE：  默认是至少一次
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22