Flink消费kafka数据起始offset配置:Flink读取Kafka数据确定开始位置有以下几种设置方式：

1. flinkKafkaConsumer.setStartFromEarliest():从topic的最早offset位置开始处理数据，如果kafka中保存有消费者组的消费位置将被忽略。
2. flinkKafkaConsumer.setStartFromLatest():从topic的最新offset位置开始处理数据，如果kafka中保存有消费者组的消费位置将被忽略。
3. flinkKafkaConsumer.setStartFromTimestamp(…):从指定的时间戳（毫秒）开始消费数据，Kafka中每个分区中数据大于等于设置的时间戳的数据位置将被当做开始消费的位置。如果kafka中保存有消费者组的消费位置将被忽略。
4. flinkKafkaConsumer.setStartFromGroupOffsets():默认的设置。根据代码中设置的group.id设置的消费者组，去kafka中或者zookeeper中找到对应的消费者offset位置消费数据。如果没有找到对应的消费者组的位置，那么将按照auto.offset.reset设置的策略读取offset。

Flink消费kafka数据，消费者offset提交配置：

配置offset的提交方式取决于是否为job设置开启checkpoint。可以使用env.enableCheckpointing(5000)来设置开启checkpoint。5000单位毫秒,代表每5秒进行依次checkpoint

• 关闭checkpoint：如何禁用了checkpoint，那么offset位置的提交取决于Flink读取kafka客户端的配置，enable.auto.commit ( auto.commit.enable【Kafka 0.8】)配置是否开启自动提交offset, auto.commit.interval.ms决定自动提交offset的周期。
• 开启checkpoint：如果开启了checkpoint，那么当checkpoint保存状态完成后，将checkpoint中保存的offset位置提交到kafka。这样保证了Kafka中保存的offset和checkpoint中保存的offset一致，可以通过配置setCommitOffsetsOnCheckpoints(boolean)来配置是否将checkpoint中的offset提交到kafka中（默认是true）。如果使用这种方式，那么properties中配置的kafka offset自动提交参数enable.auto.commit和周期提交参数auto.commit.interval.ms参数将被忽略。

总结:Flink提供了消费kafka数据的offset如何提交给Kafka或者zookeeper(kafka0.8之前,因为0.8之前offset是维护在zookeeper中的)的配置 ;关闭checkpoint的话,flink消费kafka数据 offset取决于kafka客户端的配置;开启checkpoint的话,flink消费kafka offset由jobmanager中的checkpoint维护,并同步到kafka中保持一置,注意，Flink并不依赖提交给Kafka或者zookeeper中的offset来保证容错。提交的offset只是为了外部来查询监视kafka数据消费的情况。

使用checkpoint + 两阶段提交来保证仅消费一次kafka中的数据：

Flink checkpoint机制: 这种机制是在Flink应用内部实现仅一次处理数据的基础。

当谈及“exactly-once semantics”仅一次处理数据时，指的是每条数据只会影响最终结果一次。Flink可以保证当机器出现故障或者程序出现错误时，也没有重复的数据或者未被处理的数据出现，实现仅一次处理的语义。

checkpoint中包含：

1. 当前应用的状态;
2. 当前消费流数据的位置;

注意:checkpoint机制仅限于Flink架构内部保证仅一次处理数据;

使用两阶段提交协议保证flink连接外部系统数据仅一次处理;
当Flink处理完的数据需要写入外部系统时，不保证仅一次处理数据。为了提供端到端的仅一次处理数据，在将数据写入外部系统时也要保证仅一次处理数据，这些外部系统必须提供一种手段来允许程序提交或者回滚写入操作，同时还要保证与Flink的checkpoint机制协调使用,在分布式系统中协调提交和回滚的常见方法就是两阶段提交协议。下面给出一个实例了解Flink如何使用两阶段提交协议来实现数据仅一次处理语义。

该实例是从kafka中读取数据，经过处理数据之后将结果再写回kafka。kafka0.11版本之后支持事务，这也是Flink与kafka交互时仅一次处理的必要条件。【注意：当Flink处理完的数据写入kafka时，即当sink为kafka时，自动封装了两阶段提交协议】。Flink支持仅一次处理数据不仅仅限于和Kafka的结合，只要sink提供了必要的两阶段协调实现，可以对任何sink都能实现仅一次处理数据语义。
其原理如下：

上图Flink程序包含以下组件：

1. 一个从kafka中读取数据的source
2. 一个窗口聚合操作
3. 一个将结果写往kafka的sink。
   

要使sink支持仅一次处理数据语义，必须以事务的方式将数据写往kafka,将两次checkpoint之间的操作当做一个事务提交，确保出现故障时操作能够被回滚。假设出现故障，在分布式多并发执行sink的应用程序中，仅仅执行单次提交或回滚事务是不够的，因为分布式中的各个sink程序都必须对这些提交或者回滚达成共识，这样才能保证两次checkpoint之间的数据得到一个一致性的结果。Flink使用两阶段提交协议(pre-commit+commit)来实现这个问题。

Filnk checkpointing开始时就进入到pre-commit阶段，具体来说，一旦checkpoint开始，Flink的JobManager向输入流中写入一个checkpoint barrier将流中所有消息分隔成属于本次checkpoint的消息以及属于下次checkpoint的消息，barrier也会在操作算子间流转，对于每个operator来说，该barrier会触发operator的State Backend来为当前的operator来打快照。如下图示：

Flink DataSource中存储着Kafka消费的offset，当完成快照保存后，将chechkpoint barrier传递给下一个operator。这种方式只有在Flink内部状态的场景是可行的，内部状态指的是由Flink的State Backend管理状态，例如上面的window的状态就是内部状态管理。只有当内部状态时，pre-commit阶段无需执行额外的操作，仅仅是写入一些定义好的状态变量即可，checkpoint成功时Flink负责提交这些状态写入，否则就不写入当前状态。

但是，一旦operator操作包含外部状态，事情就不一样了。我们不能像处理内部状态一样处理外部状态，因为外部状态涉及到与外部系统的交互。这种情况下，外部系统必须要支持可以与两阶段提交协议绑定的事务才能保证仅一次处理数据。

本例中的data sink是将数据写往kafka，因为写往kafka是有外部状态的，这种情况下，pre-commit阶段下data sink 在保存状态到State Backend的同时，还必须pre-commit外部的事务。如下图：

当checkpoint barrier在所有的operator都传递一遍切对应的快照都成功完成之后，pre-commit阶段才算完成。这个过程中所有创建的快照都被视为checkpoint的一部分，checkpoint中保存着整个应用的全局状态，当然也包含pre-commit阶段提交的外部状态。当程序出现崩溃时，我们可以回滚状态到最新已经完成快照的时间点。

下一步就是通知所有的operator，告诉它们checkpoint已经完成，这便是两阶段提交的第二个阶段：commit阶段。这个阶段中JobManager会为应用中的每个operator发起checkpoint已经完成的回调逻辑。本例中，DataSource和Winow操作都没有外部状态，因此在该阶段，这两个operator无需执行任何逻辑，但是Data Sink是有外部状态的，因此此时我们需要提交外部事务。如下图示：

汇总以上信息，总结得出：

1. 一旦所有的operator完成各自的pre-commit,他们会发起一个commit操作。

2. 如果一个operator的pre-commit失败，所有其他的operator 的pre-commit必须被终止，并且Flink会回滚到最近成功完成的checkpoint位置。

3. 一旦pre-commit完成，必须要确保commit也要成功，内部的operator和外部的系统都要对此进行保证。假设commit失败【网络故障原因】，Flink程序就会崩溃，然后根据用户重启策略执行重启逻辑，重启之后会再次commit。

因此，所有的operator必须对checkpoint最终结果达成共识，即所有的operator都必须认定数据提交要么成功执行，要么被终止然后回滚。

Flink中外部状态实现两阶段提交:

Flink外部状态实现两阶段提交将逻辑封装到TwoPhaseComitSinkFunction类中，下面扩展TwoPhaseCommitSinkFunction来实现就文件的sink。若要实现支持exactly-once语义的文件sink,需要实现以下4个方法：

1. beginTransaction:开启一个事务，创建一个临时文件，将数据写入到临时文件中

2. preCommit:在pre-commit阶段，flush缓存数据到磁盘，然后关闭这个文件，确保不会有新的数据写入到这个文件，同时开启一个新事务执行属于下一个checkpoint的写入操作

3. commit：在commit阶段，我们以原子性的方式将上一阶段的文件写入真正的文件目录下。【注意：数据有延时，不是实时的】

4. abort:一旦异常终止事务，程序如何处理。这里要清除临时文件。

代码如下:此次消费kafka的数据不再写往kafka

该sink必须继承 TwoPhaseCommitSinkFunction