一、简介

Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行状态计算。实现处理的低延迟、高吞吐、准确性和容错性。

需要解决的问题：
1.如何保证处理结果准确性
2.如何保证数据的时序性
3.如何保证容灾

那些行业需要处理流数据：
1.电商和市场营销：数据报表、广告投放、业务流程需要
2.物联网：传感器实时数据采集和显示、实时报警，交通运输业
3.电信业：基站流量调配
4.实时结算和通知推送，实时检测异常行为

二、处理架构对比

2.1 离线处理架构

存储与处理是分离的。

2.2 流式处理框架

2.2.1 第一代流式处理框架（Storm）

为了解决内存中数据的可靠性，引入了远程存储的概念，即有状态的流式处理。

与第三代流式处理框架

对比点	Storm	Spark Streaming
实时计算模型	纯实时，来一条数据，处理一条数据	准实时，对一个时间段内的数据收集起来，作为一个RDD，再处理
实时计算延迟度	毫秒级	秒级
吞吐量	低	高
事务机制	支持完善	支持，但不够完善
健壮性 / 容错性	ZooKeeper，Acker，非常强	Checkpoint，WAL，一般
动态调整并行度	支持	不支持

storm保证了低延迟，但是没有保证数据的吞吐量、时序性和准确性

2.2.2 第二代流式处理架构（lambda架构）

使用两套系统，流处理系统保证低延迟，批处理系统来校准结果准确性。

2.2.3 第三代流式处理架构（Flink/Spark Streaming）

Flink的主要特点：

事件驱动（Event-driven）：一个事件记录进行一次处理，区别于顺序和流程驱动。
基于流的世界观：离线数据是有界的流；实时数据是一个没有界限的流，这就是所谓的有界流和无界流。
分层API：越顶层越抽象，表达含义越简明，使用越方便；越底层越具体，表达能力越丰富，使用越灵活。如果顶层API不够用，可以通过底层API进行实现。

SQL/Table API：如果业务逻辑比较简单，可以使用顶层的API进行实现；
DataStream API：可以对流进行自定义转换和操作，如开窗；也可以使用DataSet进行批处理操作。是最为常用的API层级。
ProcessFunction（events，state，time）：对于非常复杂的业务场景，DataStream API都不能实现，那么可以使用这个最底层的API，可以自定义任何功能。可以获取当前所有的时间(state)和状态(time)，可以定义定时器。
其他特点：支持事件时间和处理时间；精确一次的状态一致性保证；低延迟，每秒处理百万个事件，毫秒级延迟；与众多常用的存储系统的连接；高可用，动态扩展，实现7*24小时全天候运行。

Spark和Flink的主要差别就在于计算模型不同。Spark采用了微批处理模型，而Flink采用了基于操作符的连续流模型。因此，对Apache Spark和Apache Flink的选择实际上变成了计算模型的选择，而这种选择需要在延迟、吞吐量和可靠性等多个方面进行权衡。
如果对流数据进行处理，推荐使用Flink，主（显而）要（易见）的原因为：

Flink灵活的窗口
Exactly Once语义保证

三、Flink的简单使用

依赖

    <properties>
        <flink.version>1.12.0</flink.version>
        <scala.binary.version>2.11</scala.binary.version>
        <slf4j.version>1.7.30</slf4j.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${slf4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-to-slf4j</artifactId>
            <version>2.14.0</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- 该插件用于将scala代码编译为class字节码文件 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>4.4.0</version>
                <executions>
                    <execution>
                        <!-- 声明编译到maven的compile阶段 -->
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.3.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

3.1 Source输入

word_count功能代码
①批处理
需要导入隐式转换：import org.apache.flink.api.scala._

import org.apache.flink.api.scala.DataSet
import org.apache.flink.api.scala._

object WordCount {
  //  **_  的用法：①包中所有类②系统默认初始化③将函数不执行返回④参数占位符⑤隐藏导入的类⑥标识符⑦绝对路径**⑧case _ 不管什么值都匹配⑨case _:BigInt =>...  当后面不用该变量，不关心变量时，可以用 _ 代替。
  def main(args: Array[String]):Unit = {
    // 创建一个批处理的执行环境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    // 从文件中读取数据
    val inputPath: String = "C:\\Users\\Administrator\\Desktop\\不常用的项目\\flink_demo\\src\\main\\resources\\test.txt"
    val inputDataSet: DataSet[String] = env.readTextFile(inputPath)

    // 对数据进行转换处理统计，先分词，再按照word进行分组，最后进行聚合统计
    val resultDataSet: DataSet[(String,Int)] = inputDataSet
      .flatMap(_.split(" "))
      .map((_, 1))
      .groupBy(0)  // 以第一个元素作为key进行分组
      .sum(1)

    resultDataSet.print()
  }
}

②流处理
如在192.168.32.242节点上启动命令nc -lk 7777向端口7777发送数据包
配置程序的参数--host 192.168.32.242 --port 7777来监听该端口获取流数据。

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._

object StreamWordCount {
  def main(args: Array[String]): Unit = {
    // 创建一个流处理的执行环境  DataStreamApi
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(8) // 设置最大并行度，默认是本机核心数

    // 接收一个socket文本流
    val paramTool: ParameterTool = ParameterTool.fromArgs(args)
    val host: String = paramTool.get("host")
    val port: Int = paramTool.getInt("port")
    val inputDataStream: DataStream[String] = env.socketTextStream(host, port)

    // 进行转化处理统计
    val resultDataStream: DataStream[(String, Int)] = inputDataStream
      .flatMap(_.split(" "))
      .filter(_.nonEmpty)
      .map((_, 1))
      .keyBy(0) // 流处理没有groupBy，使用keyBy进行聚合
      .sum(1)

    resultDataStream.print() //分布式处理会导致输出是乱序的
    resultDataStream.print().setParallelism(1) //打印的并行度是1，不会输出进程号。

    // 定义任务后，开始执行
    env.execute("stream word count")
  }
}

结果如下，最开始的数字表示运行在哪个线程下，线程号是根据key的hash值来决定的

6> (word,1)
3> (hello,1)
3> (hello,2)
5> (world,1)

原理：会将key计算得到哈希后发送到对应的线程中，这样相同key的数据发送到同一个线程中进行计算，保证了数据的准确性和吞吐量。
问题：分布式会导致时序错乱，如print()算子在不同节点上执行导致输出结果时间错乱。

③从kafka中读取流数据
添加kafka连接依赖

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

代码

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.scala._

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

    // 1.从集合中读取数据
    val readProperties = new Properties()
    readProperties.setProperty("bootstrap.servers", "192.168.32.242:9092")
    readProperties.setProperty("group.id", "consumer-group")
    readProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    readProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    val myConsumer = new FlinkKafkaConsumer[String]("kafka_test", new SimpleStringSchema(), readProperties)
    //    myConsumer.setStartFromEarliest()
    myConsumer.setStartFromLatest()

    val inputStream = env.addSource(myConsumer)

    val resultDataStream: DataStream[(String, Int)] = inputStream
      .flatMap(_.split(" "))
      .filter(_.nonEmpty)
      .map((_, 1))
      .keyBy(0) // 流处理没有groupBy，使用keyBy进行聚合
      .sum(1)

    resultDataStream.print() //分布式处理会导致输出是乱序的
    resultDataStream.print().setParallelism(1) //打印的并行度是1，不会输出进程号。

    env.execute("kafka_source_test")
  }
}

在kafka中创建topic：bin/kafka-topics.sh --zookeeper bigdata1:2181 --create --topic kafka_test --partitions 3 --replication-factor 1
启动程序，然后向创建的topic中添加数据bin/kafka-console-producer.sh --broker-list bigdata1:9092 --topic kafka_test进行测试。

④读取自定义数据源的流式数据

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.source.SourceFunction

import scala.util.Random

object SourceTest {
  // 自定义数据源读取流数据
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val stream2 = env.addSource(new MySensorSource())

    stream2.print()

    env.execute("diy_source_test")

  }

}

// 自定义函数实现run和cancel方法
class MySensorSource() extends SourceFunction[SensorReading] {
  // 定义一个标志位表示数据源是否正常发出数据
  var running: Boolean = true

  override def cancel(): Unit = running = false

  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    // 定义一个循环，不停生产数据，除非被cancel
    val curTemp = 1.to(10).map(i => ("sensor_" + i, Random.nextDouble() * 100))

    while (running) {

      curTemp.map(
        data => (data._1, data._2 + Random.nextGaussian())
      )
      val curTime = System.currentTimeMillis()
      curTemp.foreach(
        data => sourceContext.collect(SensorReading(data._1, data._2, curTime))
      )

      Thread.sleep(1000)
    }
  }

}

case class SensorReading(name: String, temp: Double, timestamp: Long)

3.2 计算API

3.3 Sink输出

输出包括文件、Kafka、ES、MySQL等

3.4 窗口API

窗口包括滚动窗口（TumblingWindow）、滑动窗口（SlidingWindow）、会话窗口（SessionWindow）和全局窗口（GlobalWindow）。

KeyedStream有window方法调用窗口函数，
DataStream有windowAll方法调用窗口函数（因为并行度是1，性能不高，所以先进行聚合），
调用窗口函数之后，得到的是WindowedStream，无界流变成了有界流。
调用聚合函数后得到的还是DataStream。

窗口函数分为增量聚合函数（数据来一条计算一次）和全窗口函数（收集后统一计算）。ReduceFunction、AggregateFunction等就是增量聚合函数，ProcessWindowFunction就是全窗口函数。

3.5 其他API

trigger() 触发器：定义window什么时候关闭，触发计算并输出结果
evictor() 移除器：定义删除某些数据逻辑
allowedLateness()：允许处理迟到的数据
sideOutputLateData()：将迟到的数据放入到侧输出流
getSideOuntput()：获取侧输出流

四、服务器中运行程序

4.1单机模式

4.1.1 启动单机模式
下载地址：https://www.apache.org/dyn/closer.lua/flink/flink-1.12.0/flink-1.12.0-bin-scala_2.12.tgz
启动命令

/bin/start-cluster.sh

启动单节点模式。
可以访问UI界面http://192.168.32.242:8081

4.1.2 提交任务

设置并行度的优先级(由高到低)

算子设置
全局设置
提交时UI界面输入设置
配置文件设置

4.1.3 上传文件设置参数

4.1.4 查看执行计划

4.1.5 启动任务
可以通过UI界面提交任务，也可以通过命令行提交任务：

bin/flink run -c com.iotmars.wecook.StreamWordCount -p 2 /opt/jar/flink-demo-0.0.1-SNAPSHOT-jar-with-dependencies.jar --host localhost --port 6666

4.1.6 查看任务

bin/flink list [-a]

4.1.7 取消任务
可以通过UI界面取消任务，也可以通过命令行取消任务：

bin/flink cancel 4198899a1a47f496309fe2da2e31c1f5

4.1.8 停止flink集群

/bin/stop-cluster.sh

4.2 Yarn模式

4.2.1 Session-cluster

概念：Session-Cluster模式需要先启动集群，然后再提交作业，接着会向yarn申请一块空间后，资源永远保持不变。如果资源满了，下一个作业无法提交，只能等到yarn中的一个作业执行完成之后释放了资源，才能进行下一个作业任务。
所有作业共享Dispatcher和ResourceManager，共享资源，适合规模小执行时间短的作业。
操作：
①启动yarn-session：

// 添加环境变量
export HADOOP_CLASSPATH=`hadoop classpath`
// 启动yarn-session
yarn-session.sh -n 2 -s 2 -jm 1024 -nm test -d
// 或调度器中创建了多个队列，需要指定队列
nohup ./yarn-session.sh -s 2 -jm 1024 -tm 2048 -nm flink-on-yarn -qu flink -d 1>/opt/module/flink-1.12.0/yarn-session.log 2>/opt/module/flink-1.12.0/yarn-session.err &

关闭yarn-session：

# 找到flink集群任务的id，然后kill
yarn application -kill application_1616059084025_0002

②提交任务(和standalone模式一样)：

bin/flink run -c com.iotmars.wecook.StreamWordCount -p 2 /opt/jar/flink-demo-0.0.1-SNAPSHOT-jar-with-dependencies.jar --host localhost --port 6666

③查看任务状态：去yarn控制台查看任务状态
④取消yarn-session：

yarn application --kill job_id

可以通过http://192.168.32.243:37807访问Web页面（会在某台服务器上部署一个Web页面）

4.2.2 Per-Job-Cluster

概念：一个Job会对应一个集群，每提交一个作业会根据自身的情况，都会单独向yarn申请资源，知道作业执行完成，一个作业失败与否不会影响下一个作业的正常提交和运行。独享Dispatcher和ResourceManager，按需接受资源申请，适合大规模长时间运行的作业。
每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成后创建的集群也会消失。
操作：
①不启动yarn-session，直接执行job

flink run `-m yarn-cluster` -c com.iotmars.wecook.StreamWordCount  /opt/jar/flink-demo-0.0.1-SNAPSHOT-jar-with-dependencies.jar --host localhost --port 7777

4.3 Kubernetes部署

Flink在最近的版本中也支持了k8s部署模式。
①搭建k8s集群
②配置各组件的yaml文件
在k8s上构建Flink Session Cluster，需要将Flink集群的组件对应的docker镜像分别在k8s上启动，包括JobManager、JobManager、JobManagerService三个镜像服务。每个镜像服务都可以从中央镜像仓库中获取。
③启动Flink Session Cluster

// 启动jobmanager-service服务
kubectl create -f jobmanager-service.yaml
// 启动jobmanager-deployment服务
kubectl create -f jobmanager-deployment.yaml
// 启动taskmanager-deployment服务
kubectl create -f taskmanager-deployment.yaml

④访问Flink UI页面
集群启动后，就可以通过JobManagerServices中配置的WebUI端口进行访问
http://{JobManagerHost:Port}/api/v1/namespaces/default/services/flink-jobmanager:ui/proxy

五、运行时流程

5.1 Flink的单机模式

Dispatcher提供了restful风格的接口用于提交任务；
JobManager分析任务生成执行图，然后向ResourceManager申请slot资源；
ResourceManager启动TaskManager，TaskManager在启动会向ResourceManager进行注册slot；
JobManager提交要在slot中执行的任务。

5.2 Yarn的Per-Job-Cluster模式下的任务提交流程

流程概述：
提交Job请求后由Yarn启动所需资源数量的容器，容器中启动的TaskManager节点向ResourceManager进行slots的注册，然后JobManager向slots分发任务执行。
相当于每次提交Job后都启动一个Flink的TaskManager集群进行处理。

流程详情：

提交JOB后，任务到YARN的ResourceManager；
RM启动AM，在AM中启动Flink的JobManager和ResourceManager；
在JobManager中进行任务切分，然后向Flink的ResourceManager进行任务请求，再向Yarn的ResourceManager进行资源申请；
Yarn的ResourceManager根据请求的资源数启动NodeManager，然后在容器中启动TaskManager；
TaskManager向Flink的ResourceManager进行slots资源注册，然后由Flink的JobManager进行资源分配。

5.3 TaskManager和Slot

两者关系

一个TaskManager至少有一个Slot，有多少个slot可以接收多少个任务。
Slot拥有独立的内存，而不是独立的cpu。可以理解为Flink中每一个TaskManager都是一个JVM进程，它会在独立的线程上执行一个或多个子任务。

任务分配

并不是每个算子的任务都会放到不同的Slot中，而是同一个任务会根据并发分配到不同的Slot中，而不同任务会分配到同一个Slot中，所以可能会有slot会保存一个pipeline(即包含了所有的算子任务)。好处是如果算子工作量不同，不会导致某些Slot闲置，可以动态分配任务量，可以充分利用多核CPU并发）。
因此所需Slot的最大数量就是任务最大并行度的值。

数据传输形式

one-to-one：map、filter、flatMap等算子符合one-to-one对应关系。这些算子的子任务得到的元素的个数和顺序跟source算子的子任务生产的元素的个数、顺序相同。
redistributing：如果前后任务并行度不同，会采用round-robin重新分区；如果是keyBy等类似shuffle的宽依赖操作算子，会进行hashCode重分区；而broadcast和rebalance会随机重分区；
local forward(本地转发)：flink采用了任务链优化技术，采用one-to-one操作且设置算子的并行度相同，可以实现本地转发进行任务连接。这样数据在本地进行流转，减少了序列化和网络传输的花销。

任务链拆分和Slot独占
如上所述，flink会自动优化任务链合并多个任务为一个任务。

如果某个算子任务特别复杂需要单独拆出来，可以在算子后面设置.disableChaining()，这样这个任务就不会被合并；
如果想把合并的任务拆分为两个任务，可以在需要拆分的算子后添加.startNewChain()。

共享组
可以将不同的任务分配到指定的共享组内。只有在同一组的算子任务会共享Slot，不同组的任务不会存在同一个Slot中。这样需要重新计算需要的Slot的数量。

如拆分出来的算子任务如果想独占一个Slot，可以在算子后面添加.slotSharingGroup("a")。

六、时间语义和水位线

6.1 时间概念

Event Time：事件创建时间
Ingestion Time：数据进入Flink的时间
Processing Time：执行操作算子的本地系统时间，与机器相关

6.2 设置时间语义

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
如果不设置，默认是Processing Time语义。

6.3 水位线

6.3.1 概念

水位线用于标记当前任务的时间，被水位线没过的时间窗口才会触发执行。当前水位线是到达的最大事件时间戳减去设置的延迟时间。

6.3.2 水位线上下游传递

任务会记录所有上游任务传递的水位线，并取最低的水位线传递给下游的任务。

当前事件时间-水位线延迟时间=水位线
水位线-窗口允许延迟时间=窗口关闭时间
注意点：

1.如果水位线没过窗口，那么会对窗口进行计算但不关闭，延迟数据进入后会重复触发窗口的计算，直到到达窗口关闭时间。
2.如果迟到数据属于多个窗口，每个窗口都会进行计算。
3.只有当迟到数据不属于任何一个窗口时，才会进入侧输出流。

水位线生成逻辑：
①
new BoundedOutOfOrdernessTimestampExtractorModelLogQ6 {
override def extractTimestamp(element: ModelLogQ6): Long = element.timestamp
}
如上对象为周期性生成watermark逻辑，默认为每隔200ms生成一个。

// 手动设置时间间隔为1s
env.getConfig().setAutoWatermarkInterval(1000);

②
new AssignerWithPunctuatedWatermarks[ModelLogQ6] {
override def checkAndGetNextWatermark(lastElement: ModelLogQ6, extractedTimestamp: Long): Watermark = ???

    override def extractTimestamp(element: ModelLogQ6, recordTimestamp: Long): Long = ???

}
每条数据到达都可以生成一个watermark。

七、Flink中的状态

算子状态（Operator state）

键控状态（keyed state）

7.2 状态后端

MemoryStateBackend：将键控状态存储在taskManager的JVM堆上，checkpoint存储在JobManager的内存中。一般测试时使用。
FsStateBackend：键控状态存储在TaskManager的JVM堆上，而将checkPoint存储在dfs上。Flink默认的配置。
RocksDBStateBackend：所有状态序列化后存储到本地RocksDB中。

7.3 一致性检查点checkpoint

checkpoint策略配置
env.enableCheckpointing(1000L) // 设置checkpoint的间隔时间
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE) //设置checkpoint的模式
env.getCheckpointConfig.setCheckpointTimeout(60000L) //设置checkpoint的超时时间
env.getCheckpointConfig.setMaxConcurrentCheckpoints(2) //允许最多两个checkpoint同时执行
env.getCheckpointconfig.setMinPauseBetweenCheckpoints(500L) //保证两次checkpoint之间至少有500ms的间隔时间
env.getCheckpointconfig.setPreferCheckpointForRecovery(true) //是否直接使用checkpoint进行故障恢复。默认是false，表示如果savepoint更近，则会使用savepoint进行恢复
env.getCheckpointconfig.setTolerableCheckpointFailureNumber(0) // 能容忍多少次的checkpoint失败。如果是0，则checkpoint失败会导致任务失败，进行重启。

重启策略配置
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,10000L)) // 固定时间间隔尝试的重启次数。
env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.of(5,TimeUnit.MINUTES)，Time.of(10,TimeUnit.SECONDS))) // 在固定时间内间隔多少时间重启多少次

7.4 保存点savepoints

原理同checkpoint，但是需要手动触发savepoint操作。
可以用于手动备份、更新应用程序、版本迁移、暂停和重启应用。
最好在程序中定义这个任务的uid(.uid("1"))，这样可以清楚找到任务对应的savepoint。

7.5 状态一致性

AT-MOST-ONCE 最多一次
AT-LEASE-ONCE 至少一次
EXACTLY-ONCE 精确一次

内部保证 —— checkpoint
source端 —— 可重设数据的读取位置
sink端 —— 从故障恢复时，数据不会重复写入外部系统

7.6 端到端一致性

7.6.1幂等写入**

7.6.2事务写入

概念：

预写日志WAL：将结果先进行保存，收到checkpoint完成通知后一次性写入到sink中。优点是简单；缺点是延迟变大而且可能会丢数据。
两阶段提交2PC：每个checkpoint，sink任务会启动一个事务，将接下来所有接收的数据添加到事务中。然后将数据写入到外部sink系统，但不提交（预提交），当收到checkpoint完成通知后才正式提交事务。Flink提供了TwoPhaseCommitSinkFunction接口。优点：实现exactly-once；缺点：需要提供事务支持的外部sink系统，而且可以在进程失败后恢复事务。如果sink超时关闭也会导致预提交的数据丢失。

Kafka中的两阶段提交
source消费Kafka数据，在checkpoint点插入barrier并记录offset到状态后端，sink会开启事务对到达的数据进行预提交。当barrier到达sink时，会开启一个新的事务对后续的数据进行预提交，直到barrier到达所有的sink，完成checkpoint后JobManager会通知Kafka进行正式提交关闭事务。
注意：1.Kafka隔离级别设置为read-commit；2.Kafka超时时间默认为15m，flink默认的checkpoint超时时间为1h。设置checkpoint超时间小于Kafka。

八、Table API 和 FLink SQL

依赖

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <!-- 引入格式描述器FormatDescriptor，适配Kafka -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>1.10.1</version>
        </dependency>

简单使用

object TableApiTest {
  def main(args: Array[String]): Unit = {
    // 1. 创建环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val tableEnv = StreamTableEnvironment.create(env)

    // 1.1 基于老版本planner的流处理
    val oldStreamSettings = EnvironmentSettings.newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val oldStreamTableEnv = StreamTableEnvironment.create(env, oldStreamSettings)

    // 1.2 基于老版本planner的批处理
    val batchEnv = ExecutionEnvironment.getExecutionEnvironment
    val oldBatchTableEnv = BatchTableEnvironment.create(batchEnv)

    // 1.3 基于blink planner的流处理
    val blinkStreamSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()
    val blinkStreamEnv = StreamTableEnvironment.create(env, blinkStreamSettings)

    // 1.4 基于blink planner的批处理
    val blinkBatchSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inBatchMode()
      .build()
    val blinkBatchEnv = TableEnvironment.create(blinkBatchSettings)


    // 2. 连接外部系统，读取数据，注册表
    // 2.1 读取文件,进行表注册并读取数据
    val path = "C:\\Users\\Administrator\\Desktop\\不常用的项目\\flink_demo\\src\\main\\resources\\test.txt";

    tableEnv.connect(new FileSystem().path(path))
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("name", DataTypes.STRING())
        .field("age", DataTypes.INT())
      )
      .createTemporaryTable("fileInputTable")

    val fileInputTable: Table = tableEnv.from("fileInputTable")
    fileInputTable.toAppendStream[(String,Int)].print()


    // 2.2 从Kafka读取数据，进行表注册并读取数据
    tableEnv
      .connect(new Kafka()
        .version("0.11")
        .topic("sensor")
        .property("zookeeper.connect", "192.168.32.242:2181")
        .property("bootstrap.servers", "192.168.32.242:9092")
      )
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("name", DataTypes.STRING())
        .field("age", DataTypes.INT())
      )
      .createTemporaryTable("kafkaInputTable")

    val kafkaInputTable = tableEnv.from("kafkaInputTable")
    kafkaInputTable.toAppendStream[(String,Int)].print()

    env.execute("table api test")
  }
}

表是由一个“标识符”来指定，由3部分组成：Catalog名、数据库（database）名和表名。
表包括常规表和视图View。

两种连接外部数据库的方式