"原文地址： [链接] 作者： [链接] 前言在消息处理过程中，除了Flink程序本身的逻辑（operator），我们还需要和外部系统进行交互，例如本地磁盘文件，HDFS，Kafka，Mysql等。虽然Flink本身支持Exactly-Once语义，但是对于完整的数据处理系统来说，最终呈现出来的语义 ...."

Fork me on GitHub

angel
AIQ 6 号成员
Flink kafka 大数据 • 0 回帖 • 3.1K 浏览 • 5 年前

Flink 实战：结合 Kafka 构建端到端的 Exactly-Once 处理程序

原文地址： https://my.oschina.net/u/992559/blog/1819948
作者： moyiguke

前言

在消息处理过程中，除了Flink程序本身的逻辑（operator），我们还需要和外部系统进行交互，例如本地磁盘文件，HDFS，Kafka，Mysql等。虽然Flink本身支持Exactly-Once语义，但是对于完整的数据处理系统来说，最终呈现出来的语义和外部系统是相关的。

我们先总览一下Flink不同connector的消息传递语义。

在Guarantees这一列，我们可以发现以下3种语义：

at most once : 至多一次。可能导致消息丢失。
at least once : 至少一次。可能导致消息重复。
exactly once ：刚好一次。不丢失也不重复。

语义分析

我们结合Kafka connector来介绍这3中不同的语义，以及分析它是如何产生的。

Kafka Producer的语义

Producer的at-most-once和at-least-once语义主要由“retries”控制（在callback中实现异常重发也相当于retry）。

如果配置值是一个大于0的整数，Producer在收到error的callback后，Producer将重新发送消息。考虑这一种情况，收到消息后，Broker正确的保存了消息，只是在返回ack时出现broker故障或者网络异常。这时候，producer收到error的callback，它不能确认异常原因，只能重新发送消息，这样就导致了消息重复。

如果配置值等于0，Producer在收到error的callback后，不重新发送消息。如果异常时由于broker没有正确保存消息导致，那么将导致消息丢失。

Producer的Exactly-Once语义，主要由“enable.idempotence”控制，如果该参数为true，将会确保消息最终只会被broker保存一次。同样的Producer在接收到error的callback后，它需要重发数据，只是在0.11以及更新的版本中，Producer会为每一批消息生成一个序列号，通过这个序列号Broker可以过滤重复消息。并且由于序列号是保存在topic上的，即使主分片失败了，新的broker也能知道消息是否需要过滤。这里还有一个细节需要注意，“acks”不能被设置为0或者1，因为万一主分片（leader replication）异常下线，将导致数据丢失，这样语义被破坏了。

NOTE : Kafka有两个概念很容易被混淆。一个是Durable，另一个是Message Delivery Semantics。这两个地方都存在消息丢失的可能性，但是机制完全不同。

Durable主要描述软件或者服务器故障后，数据是否仍能保留。Durable丢失消息主要是没有持久化：主分片收到数据后没有及时刷新到磁盘，副本没有及时复制以及持久化到磁盘。

Durable主要通过“acks”控制，最强的级别是“all”，在broker返回ack之前，它会确认每一个副本都已经保存了该消息。这样它能在n-1个副本宕机后，仍保留完整数据。最弱的级别是“0”，broker收到消息不确认持久化就返回，如果后续持久化失败，消息会丢失。当“acks”设置为“1”的时候，broker会确认主分片（leader replication）已经保存了消息，同时副本会主动向主分片同步，消息丢失风险较小。但是存在这种情况，消息到达主分片并且返回了success的ack，这时主分片fail并且副本未来得及同步这条消息，消息会丢失。

Message Delivery Semantics 主要是描述在消息系统中，消息实际被处理的次数。要区别这两点，可以简单的认为，Durable关注消息的持久化，Message Delivery Semantics关注消息的发送。

Kafka Consumer的语义

Consumer的at-most-once和at-least-once语义主要通过“offset”控制。offset的可配置为自动提交和手动提交。若配置“enable.auto.commit”为true，在Consumer fetch数据后，后台会自动提交offset。若配置“enable.auto.commit”为false,需要主动调用commitSync()或者commitAsync()来提交offset。

在自动提交的情形下，Consumer表现为at-most-once语义。在主动提交的情形下，根据用户对异常处理的不同，可表现为at-most-once或者at-least-once。

假设Consumer在fetch完数据后，后续的处理步骤出现了异常。

如果offset是自动提交的，那么Consumer将不能再次消费这些数据（除非重启Consumer，并通过seek(TopicPartition, long)重置offset）。它表现出at-most-once语义。

在捕获异常后，如果手动提交offset，表现出at-most-once语义。如果不提交offset，Consumer可重复消费该消息，表现出at-least-once语义。

在Consumer中，没有配置可以保证Exactly-Once语义。若要达到这个目标，需要在at-least-once的基础上实现幂等。这点和Producer是类似的，区别是Consumer的幂等性需要用户自己来完成。

编码准备

前面的篇幅主要介绍了Kafka的3种语义（Message Delivery Semantics），通过上述内容，我们可以得出，想要Flink和Kafka达成端到端 Exactly-Once语义，首先我们需要0.11版本或者更新的Kafka 、Producer和Consumer，其次使用幂等的Producer发送数据以及实现幂等的Consumer消费。

前置条件

JDK8
Maven 3
Git
IDE
Kafka

创建基本工程

数据以及部分代码来自http://training.data-artisans.com/ 。

Flink提供了通过mvn生成的精简Flink工程的方式，使用起来非常方便。在pom文件中，也包含了shade打包的方式，因为提交到集群上运行，需要jar-with-dependencies。

mvn archetype:generate
-DarchetypeGroupId=org.apache.flink
-DarchetypeArtifactId=flink-quickstart-java
-DarchetypeVersion=1.4.2
-DgroupId=org.apache.flink.quickstart
-DartifactId=flink-java-project
-Dversion=0.1
-Dpackage=org.apache.flink.quickstart
-DinteractiveMode=false

按实际需要修改DgroupId，DartifactId，Dversion。

下载另一个依赖工程（示例代码需要），执行install

git clone https://github.com/dataArtisans/flink-training-exercises.git

cd flink-training-exercises

mvn clean install

在建立的工程中加入上一步打包的jar作为依赖

<dependency>  <groupId>com.data-artisansgroupId>  <artifactId>flink-training-exercisesartifactId>  <version>0.15.2version>  dependency>

下载样例数据

wget http://training.data-artisans.com/trainingData/nycTaxiRides.gz

wget http://training.data-artisans.com/trainingData/nycTaxiFares.gz

导入项目到IDE，编写FlinkKafkaProducerEOSDemo

public static void main(String[] args) throws Exception {

		final int maxEventDelay = 60;       // events are out of order by max 60 seconds
		final int servingSpeedFactor = 600; // events of 10 minutes are served in 1 second

		// set up streaming execution environment
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

		String	rideInput = "./taxi-data/nycTaxiRides.gz";
		String taxiRideTopicId = "taxi-ride";

		// start the data generator
		DataStream<TaxiRide> rides = env.addSource(
				new CheckpointedTaxiRideSource(rideInput, servingSpeedFactor));

		Properties properties = new Properties();
		properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");

		SerializationSchema<TaxiRide> taxiRideSerializationSchema = new TaxiRideSchema();
		rides.addSink(new FlinkKafkaProducer011<TaxiRide>(taxiRideTopicId,
				new KeyedSerializationSchemaWrapper(taxiRideSerializationSchema),
				properties,
				FlinkKafkaProducer011.Semantic.EXACTLY_ONCE // 开启Kafka EOS
		));

		env.execute("send taxi ride to kafka ");
	}

上述代码的主体逻辑是读取nycTaxiRides.gz，并将数据发往Kafka。主要使用了CheckpointedTaxiRideSource以及FlinkKafkaProducer011。接下来，说明为什么它们能达成End-To-End的Exactly-Once语义。

CheckpointedTaxiRideSource：这是一个拥有状态的文件流，它在Checkpoint的时候记录数据读取位置（相当于Kafka的offset），Flink错误恢复后会重新定位到checkpoint记录的位置，它在整个系统上表现出来的是at-least-once。考虑这样一个场景，checkpoint成功，但是某一个commit失败，原则上本次所有的提交都要回滚。如果后续的Sink处理不当或者不支持回滚，这些数据会被提交到Sink中。在Flink 恢复后，这部分数据被重新计算，导致Sink中出现了重复的数据。

FlinkKafkaProducer011 : 提供了幂等以及事务提交。Producer的幂等性参照文章开头的语义说明，这里不再介绍。 Sink中的幂等性主要是通过两阶段提交协议来支持的（注意区分Kafka Producer本身的幂等性和依靠事务实现的幂等性）。Kafka 0.11及更新的版本提供了事务支持，可以结合Flink的两阶段提交协议使用。为了保证Sink中的数据的唯一性，将两次checkpoint之间的数据放在一个事务中，一起预提交，如果commit成功，则进入下一个checkpoint；若失败，终止事务并回滚数据。

FlinkKafkaProducer011 两阶段提交代码

protected void preCommit(FlinkKafkaProducer011.KafkaTransactionState transaction) throws FlinkKafka011Exception {
        switch(null.$SwitchMap$org$apache$flink$streaming$connectors$kafka$FlinkKafkaProducer011$Semantic[this.semantic.ordinal()]) {
        case 1:
        case 2:
            this.flush(transaction);
        case 3:
            this.checkErroneous();
            return;
        default:
            throw new UnsupportedOperationException("Not implemented semantic");
        }
    }
    protected void commit(FlinkKafkaProducer011.KafkaTransactionState transaction) {
        switch(null.$SwitchMap$org$apache$flink$streaming$connectors$kafka$FlinkKafkaProducer011$Semantic[this.semantic.ordinal()]) {
        case 1:
            transaction.producer.commitTransaction();
            this.recycleTransactionalProducer(transaction.producer);
        case 2:
        case 3:
            return;
        default:
            throw new UnsupportedOperationException("Not implemented semantic");
        }
    }
protected void abort(FlinkKafkaProducer011.KafkaTransactionState transaction) {
        switch(null.$SwitchMap$org$apache$flink$streaming$connectors$kafka$FlinkKafkaProducer011$Semantic[this.semantic.ordinal()]) {
        case 1:
            transaction.producer.abortTransaction();
            this.recycleTransactionalProducer(transaction.producer);
        case 2:
        case 3:
            return;
        default:
            throw new UnsupportedOperationException("Not implemented semantic");
        }
    }

上面是两阶段提交的主要代码。

preCommit：将本次checkpoint中未发往Broker的数据flush到Kafka Broker。这时数据已经在Kafka Broker中，但是由于事务的隔离性，Consumer暂时不会读取到这些数据（除非配置了“read_uncommitted”）。

TIPS ：为什么需要调用flush？

在Flink processElement的时候，调用KafkaProducer的send来发送数据，但是Kafka为了更高的性能，send并不立即发送数据，而是缓存在buffer中，到一定的消息量才发往Kafka Broker。这里通过flush可以强制将数据发往Kafka Broker。

commit：提交事务，这时Consumer可以读到这些数据。

abort：如果事务失败，终止事务。

FlinkKafkaConsumerEOSDemo

FlinkKafkaConsumerEOSDemo的分析流程可以参照FlinkKafkaProducerEOSDemo，这里不做细致分析。

public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); env.enableCheckpointing(100);//

    String taxiRideTopicId = "taxi-ride";
    Properties properties = new Properties();
    properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
    properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");

    DataStreamSource<TaxiRide> taxiRideDataStreamSource = env.addSource(new FlinkKafkaConsumer011<TaxiRide>(taxiRideTopicId, new TaxiRideSchema(), properties),
            "kafka-source-ride");

    String filePath = "./taxi-data/taxi-ride.txt";
    WriteFormat format = new WriteFormatAsText();
    long period = 200;
    taxiRideDataStreamSource.filter(new RideCleansing.NYCFilter()).addSink(
            new WriteSinkFunctionByMillis<TaxiRide>(filePath,format,period)
    );

    env.execute("print taxride ");


}

需要注意的是FlinkKafkaConsumer011的Exactly-Once语义通过用户配置自动设置，如果不确定Flink的语义，可以在FlinkKafkaConsumer09中打断点，断点位置：

if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS || offsetCommitMode == OffsetCommitMode.DISABLED) {
 properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
 }

自动配置相关代码：

public static OffsetCommitMode fromConfiguration(
boolean enableAutoCommit,
boolean enableCommitOnCheckpoint,
boolean enableCheckpointing) { if (enableCheckpointing) { // if checkpointing is enabled, the mode depends only on whether committing on checkpoints is enabled  return (enableCommitOnCheckpoint) ? OffsetCommitMode.ON_CHECKPOINTS : OffsetCommitMode.DISABLED;
} else { // else, the mode depends only on whether auto committing is enabled in the provided Kafka properties  return (enableAutoCommit) ? OffsetCommitMode.KAFKA_PERIODIC : OffsetCommitMode.DISABLED;
}

总结

Flink通过checkpoint和两阶段提交协议，为端到端的Exactly-Once的实现提供了可能，如果在项目中确实需要这种语义，不妨一试。

本文地址：https://www.6aiq.com/article/1545745445613
本文版权归作者和AIQ共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出