17-flink-1.10.1-flink 状态一致性

回顾

 三个策略兜底

比如写入elasticsearch 是可以做到幂等写入的，同一个_id 的值多次写入还是同一个id的值，

mysql redis 都可做到幂等写入

 5.1   如何保证

package com.study.liucf.unbounded.sink

import java.util.Properties

import com.study.liucf.bean.LiucfSensorReding
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka._

object KafkaSink {
  def main(args: Array[String]): Unit = {
    //创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置kafak配置项
    val props = new Properties()
    props.setProperty("bootstrap.servers","192.168.109.151:9092")
    props.setProperty("topic","sensor_input_csv")
    //添加kafka数据源
    val inputDs: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String]("sensor_input_csv",new SimpleStringSchema(),props))
    val transDs: DataStream[String] = inputDs.map(d=>{
      val arr = d.split(",")
      LiucfSensorReding(arr(0),arr(1).toLong,arr(2).toDouble).toString
    })
    //输出结果到标准控制台
    transDs.print()
    //输出到kafka的另一个topic里
    val outProps = new Properties()
    outProps.setProperty("bootstrap.servers","192.168.109.151:9092")
    transDs.addSink(new FlinkKafkaProducer[String](
      "sensor_out",new SimpleStringSchema(),outProps))
    //启动执行flink
    env.execute("liucf kafka sink api test")
  }
}

可以看到使用了两阶段提交

transDs.addSink(new FlinkKafkaProducer[String](
  "sensor_out",new SimpleStringSchema(),outProps))

 5.2  两阶段事务提交具体过程