原文地址: https://my.oschina.net/u/992559/blog/1819948

作者: moyiguke

前言

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

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

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

  1. at most once : 至多一次。可能导致消息丢失。
  2. at least once : 至少一次。可能导致消息重复。
  3. 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消费。

前置条件

  1. JDK8
  2. Maven 3
  3. Git
  4. IDE
  5. 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 :为什么需要�