Stream Processing with Apache Flink中文版-- 第8章 与外部系统的读写交互

数据可以存储在许多不同的系统中，比如文件系统、对象存储、关系数据库系统、键值存储、搜索索引、事件日志、消息队列等等。每一类系统都是为特定的访问模式设计的，并且擅长于服务于特定的目的。因此，今天的数据基础设施通常由许多不同的存储系统组成。在向架构中添加新组件之前，应该问一个合乎逻辑的问题:“它与架构中的其他组件的协作情况如何?”

添加数据处理系统(如Apache Flink)需要仔细考虑，因为它不包括存储层，而是依赖外部存储系统来获取和保存数据。因此，对于像Flink这样的数据处理引擎来说，为从外部系统读取数据和向外部系统写入数据，提供一个装备良好的连接器库以及实现定制连接器的API是非常重要的。但是，仅能够将数据读写到外部数据存储库对于流处理引擎来说是不够的，流处理引擎希望在发生故障时提供有意义的一致性保证。

在本章中，我们将讨论源和接收连接器如何影响Flink流应用程序的一致性保证，并介绍Flink用于读写数据的最流行的连接器。您将了解如何实现自定义源和接收连接器，以及如何实现向外部数据存储发送异步读或写请求的函数。

应用程序一致性保证

在“检查点、保存点和状态恢复”章节中，您了解到Flink的检查点和恢复机制定期接受应用程序状态的一致检查点。如果出现故障，应用程序的状态将从最新完成的检查点恢复并继续处理。但是，能够将应用程序的状态重置为一致的检查点，不足以为应用程序实现一致的处理保证。相反，应用程序的源和接收连接器需要与Flink的检查点和恢复机制集成，并提供某些属性来提供有意义的保证。

为了为应用程序提供精确一次的状态一致性保证，应用程序的每个源连接器都需要能够将其读位置设置为以前的检查点位置。在采取检查点时，源操作符将保持其读取位置，并在恢复期间恢复这些位置。支持读取位置检查点的源连接器的示例是基于文件的源，这些源将读取偏移量存储在文件的字节流中，或者Kafka源将读取偏移量存储在它所使用的主题分区中。如果应用程序从无法存储和重置读取位置的源连接器接收数据，则在发生故障时，应用程序可能会遭受数据丢失，并且只能提供最多一次的一致性保证。

Flink的检查点和恢复机制以及可重新设置的源连接器的组合保证了应用程序不会丢失任何数据。但是，应用程序可能会两次输出结果，因为在最后一个成功的检查点之后发出的所有结果(在恢复的情况下，应用程序退回到该检查点)将再次发出。因此，可重新设置的源和Flink的恢复机制不足以提供端到端的精确一次保证，即使应用程序状态是精确一次的。

旨在提供端到端(一次)保证的应用程序需要特殊的接收器连接器。有两种技术可以应用于不同的情况以实现精确一次的一致性保证：幂等写和事务性写。

幂等写

幂等运算可以执行多次，但只会导致一次更改。例如，重复地将相同的键值对插入到hashmap中是幂等操作，因为第一个插入操作将键值添加到map中，并且所有后续插入操作都不会更改map，因为它已经包含了键值对。另一方面，追加操作不是幂等操作，因为多次追加一个元素会导致多次追加。幂等写操作对于流应用程序来说很有趣，因为它们可以多次执行而不改变结果。因此，它们可以在一定程度上减轻由Flink的检查点机制引起的重放结果的影响。

应该注意的是，依赖幂等性的接收器来实现精确一次的应用程序必须保证重播时覆盖以前写的结果。例如，如果应用程序有一个接收器，它要将数据更新到键值存储中，则必须确保它能准确地计算用于更新的键值。此外，从sink系统读取数据的应用程序可能会在应用程序恢复期间观察到意外的结果。当重播开始时，先前发出的结果可能被先前的结果覆盖。因此,一个

使用恢复应用程序的输出的应用程序，可能会看到时间上的跳跃，例如，读取比以前更小的计数。此外，在重播过程中，流应用程序的整体结果将处于不一致的状态，因为一些结果将被覆盖，而另一些则没有。一旦重播完成，应用程序通过了先前失败的点，结果将再次保持一致。

事务性写

实现端到端一致性的第二种方法是基于事务写。这里的思想是只将这些结果(在最后一个成功的检查点之前计算过的结果)写入外部接收系统。此方法确保端到端精确一次，因为在出现故障时，应用程序将重置到最后一个检查点，并且在该检查点之后没有向接收系统发送任何结果。通过只在完成一个检查点后才写数据，事务处理方法不会遭遇幂等写的重播不一致性。但是，它增加了延迟，因为结果只有在检查点完成时才可见。Flink提供了两个构建块来实现事务性的接收连接器—一个通用的write-ahead-log (WAL)接收器和一个two-phase-commit(2PC)接收器。WAL sink将所有结果记录写入应用程序状态，并在接收到完成检查点的通知后将它们发送到sink系统。由于接收器缓冲记录在状态后端存储中，所以WAL接收器可以用于任何类型的接收器系统。然而，它并不是精确地提供精确一次保证的银弹，增加应用程序的状态大小，接收系统必须处理一个峰值的写入模式。

相反，2PC sink需要一个接收器系统，该系统提供事务支持或公开构建块以模拟事务。对于每个检查点，接收器启动一个事务并将所有接收到的记录附加到事务中，将它们写入接收器系统而不提交它们。当它收到一个检查点完成的通知时，它提交事务并实现结果持久化。该机制依赖于sink从在完成检查点之前打开的故障中恢复后，提交事务的能力。

2PC协议利用了Flink现有的检查点机制。检查点barriers是启动新事务的通知，所有操作符关于其单个检查点成功的通知是提交投票，而通知检查点成功的JobManager消息是提交事务的指令。与WAL sink相比，2PC sink可以根据sink系统和sink的实现实现精确的一次输出。此外，一个2PC sink不断写记录到sink系统，不会出现WAL sink那种峰值的写入模式。

表8-1显示了在最佳情况下可以实现的不同类型的source和sink连接器的端到端一致性保证;根据sink的实现，实际的一致性可能会较差。

	Nonresettable源	Resettable源
任意sink	最多一次	至少一次
幂等性sink	最多一次	精确一次(恢复期间暂时不一致)
WAL sink	最多一次	至少一次
2PC sink	最多一次	精确一次

内置连接器

Apache Flink提供连接器，用于从各种存储系统读取数据并将数据写入各种存储系统。消息队列和事件日志(如Apache Kafka、Kinesis或RabbitMQ)是读取数据流的常见源。在批处理为主的环境中，数据流也常常通过监视文件系统目录并在文件出现时读取它们来接收数据。

在sink端,数据流往往产生到消息队列，用于后续的事件流处理应用,写入文件系统归档或使数据可用于离线分析或批处理应用程序,或插入键值存储或关系数据库系统,如Cassandra,ElasticSearch,或MySQL，是数据可搜索和可查询,或服务于指示面板应用程序。

不幸的是，除了用于关系DBMS的JDBC外，大多数这些存储系统都没有标准接口。相反，每个系统都有自己的带有专用协议的连接器类库。因此，像Flink这样的处理系统需要维护几个专用的连接器，以便能够从最常用的消息队列、事件日志、文件系统、键值存储和数据库系统中读取事件并将事件写入其中。

Flink为Apache Kafka、Kinesis、RabbitMQ、Apache Nifi、各种文件系统、Cassandra、ElasticSearch和JDBC提供连接器。此外，Apache Bahir项目还为ActiveMQ、Akka、Flume、Netty和Redis提供了额外的Flink连接器。

为了在应用程序中使用提供的连接器，需要将其依赖项添加到项目的构建文件中。我们在“引入外部和Flink依赖项”章节中解释了如何添加连接器依赖项。

在下一节中，我们将讨论Apache Kafka、基于文件的源和sink以及Apache Cassandra的连接器。这些是最广泛使用的连接器，它们也代表了源和sink系统的重要类型。您可以在Apache Flink或Apache Bahir的文档中找到关于其他连接器的更多信息。

Apache Kafka源连接器

Apache Kafka是一个分布式流平台。它的核心是一个分布式发布-订阅消息传递系统，广泛用于接收和分发事件流。在深入研究Flink的Kafka连接器之前，我们简要地解释一下Kafka的主要概念。

Kafka将事件流组织为所谓的主题(Topic)。主题是一个事件日志，它保证事件按写入的顺序读取。为了扩展主题的写和读，可以将主题划分为分布在集群中的分区。顺序保证仅限于分区—kafka在从不同分区读取时不提供顺序保证。Kafka分区中的读取位置称为偏移量。

Flink为所有常见的Kafka版本提供源连接器。通过Kafka 0.11，客户端库的API得到了改进，并添加了新的特性。例如，Kafka 0.10增加了对记录时间戳的支持。自发布1.0以来，API一直保持稳定。Flink提供了一个通用的Kafka连接器，适用于0.11以后的所有Kafka版本。Flink还为Kafka的0.8、0.9、0.10和0.11版本提供了特定于版本的连接器。对于本节的其余部分，我们将重点讨论通用连接器，而对于特定于版本的连接器，我们建议您参考Flink的文档。

通用 Flink Kafka连接器的依赖项添加到Maven项目中，如下图所示:

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-connector-kafka_2.12</artifactId>
   <version>1.7.1</version>
</dependency>

Flink Kafka连接器并行地接收事件流。每个并行源任务可以从一个或多个分区读取数据。任务跟踪每个分区的当前读取偏移量，并将其包含到检查点数据中。从失败中恢复时，将恢复偏移量，并且源实例将继续从检查点偏移量读取数据。Flink Kafka连接器不依赖Kafka自己的偏移跟踪机制，该机制基于所谓的消费者组。图8-1显示了对源实例的分区分配。

图8-1 读取Kafka Topic分区的偏移量

创建一个Kafka源连接器，如示例8-1所示。

val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
properties.setProperty("group.id", "test")
val stream: DataStream[String] = env.addSource(
new FlinkKafkaConsumer[String]("topic",new SimpleStringSchema(),properties))

构造函数有三个参数。第一个参数定义要读取的主题。可以是单个主题、主题列表，也可以是匹配所有要读取的主题的正则表达式。当从多个主题读取时，Kafka连接器将所有主题的所有分区都视为相同的，并将它们的事件多路复用到单个流中。

第二个参数是DeserializationSchema 或 KeyedDeserializationSchema。Kafka消息存储为原始字节消息，需要反序列化为Java或Scala对象。在例8-1中使用的SimpleStringSchema是一个内置的DeserializationSchema，它只是将字节数组反序列化为字符串。此外，Flink还为Apache Avro和基于文本的JSON编码提供了实现。

DeserializationSchema 和KeyedDeserializationSchema是公共接口，因此您可以始终实现自定义的反序列化逻辑。

第三个参数是一个Properties对象，它配置用于连接和读取Kafka的Kafka客户端。一个最小的属性配置包含两个属性，"bootstrap.servers" 和"group.id"。有关其他配置属性，请参阅Kafka文档。为了获取事件时间时间戳并生成水印，可以通过调用FlinkKafkaConsumer.assignTimestampsAndWatermark()向Kafka 消费者提供一个AssignerWithPeriodicWatermark o或an AssignerWithPunctuatedWatermark。将分配程序应用于每个分区，以利用每个分区的排序保证，并且源实例根据水印传播协议合并分区水印(请参阅“水印传播和事件时间”)。

                                          注意
请注意，如果一个分区处于不活动状态(不提供消息)，则源实例的水印将不起作用。因此，一个不活动的分区会导致整个应用程序停顿，因为应用程序的水印不可用。

从0.10.0版本开始，Kafka支持消息时间戳。当从Kafka版本0.10或更高版本读取消息时，如果应用程序以事件时间模式运行，消费者将自动提取消息时间戳作为事件时间戳。在这种情况下，您仍然需要生成水印，并且应该应用AssignerWithPeriodicWatermark 或 AssignerWithPunctuatedWatermark来转发之前分配的Kafka时间戳。

还有一些需要注意的配置选项。如可以配置最初读取Topic分区的起始位置。有效的选项是:

• 对于一个消费者组而言，kafka通过group.id知道最后一次的消费位置，这也是默认配置：

FlinkKafkaConsumer.setStartFromGroupOffsets()

• 从每个分区的最开始位置消费：

FlinkKafkaConsumer.setStartFromEarliest()

• 从每个分区的最新位置消费：

FlinkKafkaConsumer.setStartFromLatest()

• 消费大于指定时间戳的所有记录(Kafka版本0.10或更高版本)：

FlinkKafkaConsumer.setStartFromTimestamp(long)

• 使用map对象，指定每个分区的消费起始位置：

FlinkKafkaConsumer.setStartFromSpecificOffsets(Map)

                                          注意
注意，这种配置只影响第一次读位置。在进行恢复或从保存点开始时，应用程序将从存储在检查点或保存点中的偏移量开始读取。

可以将Flink Kafka消费者配置为自动发现与正则表达式匹配的新Topic或添加到Topic中的新分区。这些特性在默认情况下是禁用的，可以通过向Properties对象添加具有非负值的参数flink.partitiondiscovery.interval-millis来启用。

Apache Kafka sink连接器

Flink为0.8以后的所有Kafka版本提供sink连接器。从Kafka 0.11，客户端的API得到了改进，并添加了新的特性，比如Kafka 0.10支持记录时间戳，Kafka 0.11支持事务性写。自发布1.0以来，API一直保持稳定。Flink提供了一个通用的Kafka连接器，适用于0.11以后的所有Kafka版本。Flink还提供了针对Kafka 0.8、0.9、0.10和0.11版本的特定于版本的连接器。对于本节的其余部分，我们将重点介绍通用连接器，并向您介绍Flink的文档，以获得特定于版本的连接器。Flink的通用Kafka连接器的依赖项被添加到Maven项目中，如下图所示:

<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-connector-kafka_2.12</artifactId>
   <version>1.7.1</version>
</dependency>

将Kafka sink添加到DataStream应用程序中，如例8-2所示

val stream: DataStream[String] = ...
val myProducer = new FlinkKafkaProducer[String](
               "localhost:9092", // broker list
               "topic", // target topic
               new SimpleStringSchema) // serialization schema
stream.addSink(myProducer)

例8-2中使用的构造函数接收三个参数。第一个参数是一个逗号分隔的Kafka broker地址字符串。第二个是写入数据的topic的名称，最后一个是SerializationSchema，它将sink的输入类型(例8-2中的字符串)转换为字节数组。SerializationSchema是我们在Kafka源连接器部分讨论的DeserializationSchema的对应版本。

FlinkKafkaProducer提供了更多具有不同参数组合的构造函数，如下:

与Kafka源连接器类似，可以为Kafka客户端传递一个Properties对象来提供定制选项。在使用Properties时，必须将brokers列表作为“bootstrap.servers”属性。查看Kafka文档以获得完整的参数列表。

您可以指定一个FlinkKafkaPartitioner来控制记录如何映射到Kafka分区。我们将在本节后面更深入地讨论这个特性。

您还可以指定KeyedSerializationSchema，而不是使用SerializationSchema将记录转换为字节数组，KeyedSerializationSchema将记录序列化为两个字节数组—一个用于键，另一个用于Kafka消息的值。此外，KeyedSerializationSchema还公开了更多的功能，比如覆盖目标主题以写入多个主题。

kafka sink 至少一次的保证

Flink的Kafka sink提供的一致性保证取决于它的配置。Kafka sink在以下条件下提供至少一次的保证:

• 启用了Flink的检查点，应用程序的所有sources都可以重新设置。

• 如果写操作不成功，sink连接器将抛出异常，导致应用程序失败并恢复。这是默认的行为。通过将retries属性设置为大于0的值(默认值)，可以将Kafka客户端配置为在写操作失败之前进行重试。您还可以通过在接收器对象上调用setLogFailuresOnly(true)来将接收器配置为只记录写故障。注意，这将使应用程序的输出保证无效。

• sink连接器等待Kafka在完成其检查点之前确认输出记录。这是默认的行为。通过调用sink对象上的setFlushOnCheckpoint(false)，可以禁用这种等待。但是，这也将禁用任何输出保证。

kafka sink 精确一次的保证

Kafka 0.11引入了对事务性写的支持。由于这个特性，Flink的Kafka接收器也能够提供精确的一次输出保证，只要接收器和Kafka配置正确。同样，Flink应用程序必须启用检查点并从可重置的源消费。此外，FlinkKafkaProducer提供了一个语义参数的构造函数，该参数控制接收器提供的一致性保证。可能的一致性值为:

• Semantic.NONE，它不提供任何保证——记录可能丢失或多次写入。

• Semantic.AT_LEAST_ONCE，它保证没有写操作丢失，但是可能会重复。这是默认设置。

• Semantic.EXACTLY_ONCE，它构建在Kafka的事务上，将每个记录精确地写入一次。

当使用Kafka接收器以精确一次模式运行Flink应用程序时，需要考虑一些事情，这有助于大致了解Kafka如何处理事务。简而言之，Kafka的事务将所有消息添加到分区日志中，并将打开的事务标记为未提交。一旦事务被提交，标记就被更改为已提交。从topic读取数据的消费者可以配置一个隔离级别(通过isolation.level属性)。声明它是否可以读取未提交的消息(默认的read_uncommitted)。如果消费者被配置为read_committed，那么一旦它遇到未提交的消息，它就停止从一个分区消费，并在提交消息时继续使用。因此，打开的事务可能会阻止消费者读取分区消息并带来显著的延迟。Kafka通过在超时间隔后拒绝和关闭事务来防止这种情况，超时间隔使用transaction.timeout.ms的属性进行配置。

在Flink的Kafka sink上下文中，这很重要，因为由于恢复周期过长而超时的事务会导致数据丢失。因此，正确配置事务超时属性非常重要。默认情况下，Flink Kafka sink设置transaction.timeout.ms为一小时。这意味着您可能需要调整kafka本身的transaction.max.timeout.ms属性，默认设置为15分钟。此外，提交消息的可见性取决于Flink应用程序的检查点间隔。请参阅Flink文档，了解在启用精确一次一致性时的其他一些情况。

                                    检查kafka集群的配置
Kafka集群的默认配置仍然会导致数据丢失，即使在确认写操作之后也是如此。您应该仔细修改Kafka设置的配置，特别注意以下参数:
ack
log.flush.interval.messages
log.flush.interval.ms
log.flush。*
我们建议您参考Kafka文档，以获得关于它的配置参数的详细信息和适用配置的指导原则。

自定义分区和写入消息时间戳

当将消息写入Kafka topic时，Flink Kafka sink任务可以选择写入topic的哪个分区。FlinkKafkaPartitioner可以在Flink Kafka sink的一些构造函数中定义。如果没有指定，默认的分区程序将每个sink任务映射到一个Kafka分区—----由同一个sink任务发出的所有记录都被写到同一个分区，如果任务多于分区，单个分区可能包含多个sink任务的记录。如果分区的数量大于子任务的数量，则默认配置将导致空分区，事件时间模式下的Flink应用程序，消费此时的topic是，可能会出现问题。

通过提供一个自定义FlinkKafkaPartitioner，您可以控制如何将记录路由到topic分区。例如，可以根据记录的key属性创建分区程序，或者创建循环分区程序以实现均匀分布。还可以根据消息key将分区委托给Kafka。这需要一个KeyedSerializationSchema来提取消息key，并使用null配置FlinkKafkaPartitioner参数来禁用默认分区程序。

最后，可以将Flink的Kafka sink配置为写入消息时间戳，这是Kafka 0.10所支持的。通过在sink对象上调用setWriteTimestampToKafka(true)，可以将记录的事件时间戳写入Kafka。

Filesystem源连接器

Filesystems(文件系统)通常用于以一种经济有效的方式存储大量数据。在大数据架构中，它们通常充当批处理应用程序的数据源和数据接收器。与高级文件格式(如Apache Parquet或Apache ORC)相结合，文件系统可以有效地为分析查询引擎(如Apache Hive、Apache Impala或Presto)提供服务。因此，文件系统通常用于“连接”流和批处理应用程序。

Apache Flink提供了一个可重置的源连接器，可以将文件中的数据作为流获取。文件系统源是flinkstreaming- java模块的一部分。因此，您不需要添加任何其他依赖项来使用此功能。Flink支持不同类型的文件系统，比如本地文件系统(包括本地装载的NFS或SAN共享、Hadoop HDFS、Amazon S3和OpenStack Swift FS)。请参阅“文件系统配置”以了解如何在Flink中配置文件系统。示例8-3显示了如何通过按行读取文本文件来读取流。

val lineReader = new TextInputFormat(null)
val lineStream: DataStream[String] = env.readFile[String](
lineReader, // The FileInputFormat
"hdfs:///path/to/my/data", // The path to read
FileProcessingMode.PROCESS_CONTINUOUSLY, // The processing mode
30000L) // The monitoring interval in ms

StreamExecutionEnvironment.readFile()方法的参数为:

• 负责读取文件内容的FileInputFormat。我们将在本节的后面讨论这个接口的细节。例8-3中的TextInputFormat的null参数定义了单独设置的路径。

• 应该读取的路径。如果路径引用一个文件，则读取单个文件。如果它引用一个目录，FileInputFormat将扫描该目录以查找要读取的文件。

• 读取路径的模式。该模式可以是PROCESS_ONCE 或 PROCESS_CONTINUOUSLY。在PROCESS_ONCE模式中，当作业启动并读取所有匹配的文件时，读取路径将被扫描一次。在 PROCESS_CONTINUOUSLY中，将定期扫描路径(在初始扫描之后)，并不断读取新的和修改过的文件。

• 周期性扫描路径的毫秒间隔。在PROCESS_ONCE模式中忽略该参数。

FileInputFormat是一种专门用于从文件系统中读取文件的InputFormat。FileInputFormat分两个步骤读取文件。首先，它扫描文件系统路径，并为所有匹配的文件创建所谓的输入分片。输入分片定义文件上的范围，通常通过起始偏移量和长度定义。在将一个大文件分成多个分段之后，可以将这些分段分配给多个读取器任务来并行地读取文件。根据文件的编码，可能需要只生成一个分割来读取整个文件。FileInputFormat的第二步是接收输入分割，读取分割定义的文件范围，并返回所有相应的记录。

DataStream应用程序中使用的FileInputFormat还应该实现CheckpointableInputFormat接口，该接口定义了检查点的方法，并在文件分割中重置InputFormat的当前读取位置。如果FileInputFormat没有实现CheckpointableInputFormat接口，则文件系统源连接器仅在启用检查点时至少提供一次保证，因为输入格式将从上次执行完整检查点时处理的分割开始读取。

在1.7版本中，Flink提供了一些扩展FileInputFormat和实现CheckpointableInputFormat的类。TextInputFormat按行读取文本文件(按换行字符分隔)，CsvInputFormat的子类按逗号分隔值读取文件，AvroInputFormat按avro编码记录读取文件。

在PROCESS_CONTINUOUSLY模式下，文件系统(filesystem)源连接器根据修改时间戳识别新文件。这意味着如果一个文件被修改，它将被完全重新处理，因为修改时间戳发生了变化。这包括由于追加内容而引起的修改。因此，持续读取文件的一种常见技术是将它们写入临时目录，并在完成后自动将它们移到受监控的目录中。当一个文件被完全读取并且完成了一个检查点时，就可以从目录中删除它。如果您使用最终一致的列表操作(如S3)从文件存储中读取数据，那么通过跟踪修改时间戳来监控读取的文件也会产生影响。由于文件可能不会按照修改时间戳的顺序出现，文件系统源连接器可能会忽略它们。

请注意，在PROCESS_ONCE模式中，在扫描文件系统路径并创建所有的分片之后，不会采取任何检查点。

如果你想使用一个文件系统源连接器在事件时间应用程序中,您应该清楚，生成水印会是一个挑战，因为输入分片由三个进程产生，然后轮询分发给并行读取进程。为了生成令人满意的水印，您需要对包含在稍后由任务处理的分片中的记录的最小时间戳进行推断。

Filesystem Sink 连接器

将流写入文件是一个常见的需求，例如，为离线交互分析准备低延迟的数据。由于大多数应用程序只有在文件完成写入，并流应用程序长时间运行之后才能读取文件，所以流 sink 连接器通常会将其输出分块到多个文件中。此外，将记录组织到所谓的bucket中是很常见的，这样消费应用程序可以更好地控制读取哪些数据。

与文件系统源连接器一样，Flink的StreamingFileSink连接器也包含在flink-streaming-java模块中。因此，不需要向构建文件添加依赖项。

StreamingFileSink为应用程序提供端到端的精确一次保证，前提是应用程序配置了精确一次检查点，并且在出现故障时重置所有源。我们将在本节后面更详细地讨论恢复机制。示例8-4展示了如何使用最少的配置创建StreamingFileSink并将其附加到流中。

val input: DataStream[String] = …
val sink: StreamingFileSink[String] = StreamingFileSink.forRowFormat(
   new Path("/base/path"),
   new SimpleStringEncoder[String]("UTF-8")
).build()

input.addSink(sink)

当StreamingFileSink接收到一条记录时，将该记录分配给一个bucket。bucket是基路径的一个子目录，在示例8-4中使用StreamingFileSink构建器配置了“/base/path”。

bucket是由BucketAssigner选择的，它是一个公共接口，并为每个记录返回一个BucketId，该BucketId确定记录将被写入的目录。可以使用withBucketAssigner()方法在构建器上配置BucketAssigner()。如果没有显式指定BucketAssigner，则使用DateTimeBucketAssigner，根据记录写入时的处理时间将记录分配到每小时的bucket。

每个bucket目录包含多个分片文件，这些文件由StreamingFileSink的多个并行实例并发生成。此外，每个并行实例将其输出分割成多个分片文件。分片文件的路径格式如下:

[base-path]/[bucket-path]/part-[task-idx]-[id]

例如，给定一个“/johndoe/demo”的基路径和一个part前缀“part”，这个路径“/johndoe/demo/2018-07-22-17/part-4-8”指向由第五个(下标从0开始)sink任务写入bucket“2018-07-22-17”的8个文件，即：2018年7月22日下午5点。

                                  提交文件的id可能不是连续的
非连续文件id(提交文件名称中的最后一个数字)不表示数据丢失。StreamingFileSink只是增加文件id。当丢弃挂起的文件时，它不会重用它们的id。

RollingPolicy确定任务何时创建新分片文件。可以使用构建器上的withRollingPolicy()方法来配置RollingPolicy。默认情况下，StreamingFileSink使用一个DefaultRollingPolicy，配置为当分片文件超过128 MB或超过60秒时滚动生成它们。还可以配置一个不活动的时间间隔，在此之后将滚动生成分片文件。

StreamingFileSink支持将记录写入分片文件的两种模式:row编码和buik编码。在row编码模式中，每个记录都单独编码并附加到一个分片文件中。在bulk编码中，记录是成批收集和写入的。Apache Parquet以列格式组织和压缩记录，是一种需要bulk编码的文件格式。

例8-4通过提供一个将单个记录写入分片文件的Encoder，使用row编码创建StreamingFileSink。在例8-4中，我们使用了SimpleStringEncoder，它调用了记录的toString()方法，并将记录的字符串表示形式写入文件。Encoder是一个简单的接口,只有一个方法，可以很容易地实现。

例8-5所示，创建了一个bulk编码的StreamingFileSink。

val input: DataStream[String] = …
val sink: StreamingFileSink[String] = StreamingFileSink.forBulkFormat(
   new Path("/base/path"),
   ParquetAvroWriters.forSpecificRecord(classOf[AvroPojo])
).build()

input.addSink(sink)

bulk编码模式下的StreamingFileSink需要BulkWriter.Factory。在例8-5中，我们对Avro文件使用了Parquet写入器。请注意，Parquet写入器包含在flinkparquet模块中，需要将其作为依赖项添加。像往常一样,BulkWriter.Factory是一个可以实现自定义文件格式(如Apache Orc)的接口。

                                            请注意
bulk编码模式下的StreamingFileSink不能选择RollingPolicy。bulk编码格式只能与OnCheckpointRollingPolicy相结合，OnCheckpointRollingPolicy在每个检查点上滚动生成分片文件。

StreamingFileSink提供了精确的一次输出保证。StreamingFileSink通过一个提交协议来实现这一点，该协议将文件通过不同的stages，处理中状态，挂起状态和完成状态进行移动，基于Flink的检查点机制。当sink写入文件时，文件处于处理中状态。当RollingPolicy决定滚动文件时，将关闭该文件并通过重命名将其移动到挂起状态。当下一个检查点完成时，挂起的文件将移动到完成状态(再次通过重命名)。

                                挂起的文件可能永远不会被提交
在某些情况下，永远不会提交挂起文件。StreamingFileSink确保这不会导致数据丢失。但是，这些文件不会自动清除。
在手动删除一个挂起文件之前，您需要检查它是在延迟还是即将提交。找到具有相同任务索引和更大ID的提交文件后，可以安全地删除挂起文件。

在失败的情况下，sink任务需要将当前正在处理的文件重置为最近一次成功检查点处的写偏移量。这是通过关闭当前正在处理的文件并丢弃文件末尾的无效部分来实现的，例如，通过使用文件系统的truncate操作。

                                STREAMINGFILESINK需要启用检查点
如果应用程序没有启用检查点，那么StreamingFileSink将永远不会将文件从挂起状态移动到完成状态。

Apache Cassandra Sink 连接器

略。

实现自定义源函数

DataStream API提供了两个接口来实现源连接器和相应的RichFunction抽象类:

• SourceFunction和RichSourceFunction可用于定义非并行源连接—与单个任务一起运行的源。

• ParallelSourceFunction和RichParallelSourceFunction可用于定义运行多个并行任务实例的源连接器。

除了非并行和并行之外，这两个接口是相同的。与处理函数的丰富变体一样，RichSourceFunction和RichParallelSourceFunction的子类可以覆盖open()和close()方法，并访问RuntimeContext，其中提供并行任务实例的数量和当前实例的索引。

SourceFunction 和 ParallelSourceFunction定义了两个方法：

• void run(SourceContext<T> ctx)

• void cancel()

run()方法执行读取或接收记录并将其放入到Flink应用程序中的实际工作。根据接收数据的系统，可以推或拉数据。run()方法由Flink调用一次，并在专用的源线程中运行，通常在一个无限循环(无限流)中读取或接收数据并发出记录。可以在某个时间点显式地取消任务，或者在有限流的情况下，当输入被完全消耗时终止任务。

当应用程序被取消和关闭时，Flink调用cancel()方法。为了执行适当的关闭，在单独的线程中运行的run()方法应该在调用cancel()方法时立即终止。示例8-10显示了一个简单的源函数，其计数范围从0到Long.MaxValue。

class CountSource extends SourceFunction[Long] {
   var isRunning: Boolean = true
   override def run(ctx: SourceFunction.SourceContext[Long]) = {
       var cnt: Long = -1
       while (isRunning && cnt < Long.MaxValue) {
           cnt += 1
           ctx.collect(cnt)
      }
  }
   override def cancel() = isRunning = false
}

重置源函数

在本章的前面，我们解释了Flink只能为使用源连接器的应用程序提供令人满意的一致性保证，这些源连接器可以重放它们的输出数据。如果提供数据的外部系统公开API来检索和重置读取偏移量，则源函数可以重播其输出。此类系统的示例包括提供文件流偏移量的文件系统和将文件流移动到特定位置的seek方法，或者Apache Kafka，后者为主题的每个分区提供偏移量，可以设置分区的读取位置。一个反例是一个从网络套接字读取数据的源连接器，它会立即丢弃传输的数据。

支持输出回放的源函数需要与Flink的检查点机制集成，并且必须在采取检查点时持久化当前的所有读取位置。当应用程序从保存点启动或从故障中恢复时，将从最新的检查点或保存点检索读取偏移量。如果应用程序在没有现有状态的情况下启动，则必须将读取偏移量设置为默认值。复位源函数需要实现CheckpointedFunction接口,存储读取偏移和所有相关的元数据信息,如文件路径或分区ID,在操作符列表状态或操作符 union list状态，取决于offsets应该如何分布给并行的task实例。有关操作符列表状态和union list状态的分发行为的详细信息，请参阅“缩放有状态操作符(Scaling Stateful Operators)”章节。

此外，确保在单独的线程中运行的SourceFunction.run()方法不会提前读取偏移量并在采取检查点时发出数据，这一点非常重要;换句话说，当调用CheckpointedFunction.snapshotState()方法时。这是通过保护run()中的代码来实现的，run()将读取位置提前，并在一个块中发出记录，该块在一个锁对象上进行同步，该对象是从SourceContext.getCheckpointLock()方法获得的。例8-11使例8-10的CountSource可重新设置。

class ResettableCountSource extends SourceFunction[Long] with CheckpointedFunction {
   var isRunning: Boolean = true
   var cnt: Long = _
   var offsetState: ListState[Long] = _
   override def run(ctx: SourceFunction.SourceContext[Long]) = {
       while (isRunning && cnt < Long.MaxValue) {
           // synchronize data emission and checkpoints
           ctx.getCheckpointLock.synchronized {
               cnt += 1
               ctx.collect(cnt)
          }
      }
  }
   override def cancel() = isRunning = false
   override def snapshotState(snapshotCtx:FunctionSnapshotContext): Unit = {
       // remove previous cnt
       offsetState.clear()
       // add current cnt
       offsetState.add(cnt)
  }
   override def initializeState(initCtx: FunctionInitializationContext): Unit = {
       val desc = new ListStateDescriptor[Long]("offset",classOf[Long])
       offsetState =initCtx.getOperatorStateStore.getListState(desc)
       // initialize cnt variable
       val it = offsetState.get()
       cnt = if (null == it || !it.iterator().hasNext) {
           -1L
      } else {
           it.iterator().next()
      }
  }
}

源函数、时间戳和水印

源函数的另一个重要方面是时间戳和水印。正如在“事件时间处理”和“分配时间戳和生成水印”中指出的，DataStream API提供了两个选项来分配时间戳和生成水印。时间戳和水印可以由专用的TimestampAssigner分配和生成(详细信息请参阅“分配时间戳和生成水印”)，也可以由源函数分配和生成。

源函数分配时间戳并通过其SourceContext对象发出水印。SourceContext提供了以下方法:

• def collectWithTimestamp(T record, longtimestamp): Unit

• def emitWatermark(Watermark watermark):Unit

collectWithTimestamp()输出带有相关时间戳的记录，emitWatermark()输出提供的水印。

如果源函数的一个并行实例使用来自多个流分区(例如Kafka主题的分区)的记录，那么除了不需要额外的操作符外，在源函数中分配时间戳和生成水印也是有益的。通常，外部系统(如Kafka)只保证流分区中的消息顺序。给定一个并行度为2的源函数操作符，它用6个分区从一个Kafka主题读取数据，源函数的每个并行实例将从3个Kafka主题分区读取记录。因此，源函数的每个实例多路复用三个流分区的记录来输出它们。多路复用记录很可能会在事件时间戳方面引入额外的无序性，这样下游时间戳分配者可能会产生比预期更多的延迟记录。

为了避免这种行为，源函数可以独立地为每个流分区生成水印，并且始终将其分区的最小水印作为水印。通过这种方式，它可以确保利用每个分区上的顺序保证，并且不会输出不必要的延迟记录。

源函数必须处理的另一个问题是实例变得空闲并且不再输出任何数据。这是非常有可能产生问题的，因为它可能会阻止整个应用程序前进其水印，从而导致一个停滞的应用程序。由于水印应该是数据驱动的，所以如果没有接收到输入记录，水印生成器(集成在源函数或时间戳分配程序中)将不会发出新的水印。如果您查看一下Flink是如何传播和更新水印的(请参阅“水印传播和事件时间”)，您就会发现，如果应用程序涉及到一个shuffle操作(keyBy()、rebalance()等)，那么一个不提前使用水印的操作符就可以停止应用程序的所有水印。

Flink提供了一种机制，通过将源函数标记为临时空闲来避免这种情况。当处于空闲状态时，Flink的水印传播机制将忽略空闲流分区。一旦源再次开始发出记录，它就会被自动设置为活动的。源函数可以通过调用SourceContext.markAsTemporarilyIdle()方法来决定何时将自己标记为空闲。

实现自定义接收器函数

在Flink的DataStream API中，任何操作符或函数都可以将数据发送到外部系统或应用程序。数据流最终不必流到接收器操作符中。例如，您可以实现一个FlatMapFunction，它通过HTTP POST调用而不是通过它的收集器来输出每个传入的记录。尽管如此，DataStream API提供了一个专用的SinkFunction接口和一个相应的RichSinkFunction抽象类。SinkFunction接口提供了一个单一的方法:

void invoke(IN value, Context ctx)

SinkFunction的上下文对象提供了对当前处理时间、当前水印以及记录的时间戳的访问。

例8-12显示了一个简单的SinkFunction，它将传感器读数写入socket。注意，在启动程序之前，您需要启动一个监听socket的进程。否则，由于无法打开到socket的连接，程序会因ConnectException异常而失败。在Linux上运行命令nc -l localhost 9191来监听localhost:9191。

val readings: DataStream[SensorReading] = ???
// write the sensor readings to a socket
readings.addSink(new SimpleSocketSink("localhost", 9191))
// set parallelism to 1 because only one thread can write to a socket
.setParallelism(1)
// -----
class SimpleSocketSink(val host: String, val port: Int)
extends RichSinkFunction[SensorReading] {
   var socket: Socket = _
   var writer: PrintStream = _
   override def open(config: Configuration): Unit = {
       // open socket and writer
       socket = new Socket(InetAddress.getByName(host), port)
       writer = new PrintStream(socket.getOutputStream)
  }
   override def invoke(
       value: SensorReading,
       ctx: SinkFunction.Context[_]): Unit = {
       // write sensor reading to socket
       writer.println(value.toString)
       writer.flush()
  }
   override def close(): Unit = {
       // close writer and socket
       writer.close()
       socket.close()
  }
}

如前所述，应用程序的端到端精确一致性保证取决于其sink连接器的属性。为了实现端到端的精确一次语义，应用程序需要幂等或事务接收连接器。例8-12中的SinkFunction既不执行幂等写，也不提供事务性写。由于套接字的仅提供追加特性，因此无法执行幂等写操作。由于套接字没有内置的事务支持，所以只能使用Flink的通用WAL sink完成事务写。在接下来的部分中，您将了解如何实现幂等或事务接收连接器。

幂等sink连接器

对于许多应用程序，SinkFunction接口足以实现幂等接收器连接器。这是可能的，如果满足以下两点:

1. 结果数据具有一个确定性(复合)key，可以对其执行幂等更新。对于计算每个传感器和分钟的平均温度的应用程序，确定性key可以是传感器的ID和每分钟的时间戳。确定性key对于确保在发生恢复时正确地覆盖所有写操作非常重要。

2. 外部系统支持每个键的更新，比如关系数据库系统或键值存储。

示例8-13说明了如何实现和使用向JDBC数据库(在本例中是嵌入式Apache Derby数据库)写入的幂等SinkFunction。

val readings: DataStream[SensorReading] = ???
// write the sensor readings to a Derby table
readings.addSink(new DerbyUpsertSink)
// -----
class DerbyUpsertSink extends RichSinkFunction[SensorReading] {
var conn: Connection = _
var insertStmt: PreparedStatement = _
var updateStmt: PreparedStatement = _
override def open(parameters: Configuration): Unit = {
   // connect to embedded in-memory Derby
   conn = DriverManager.getConnection("jdbc:derby:memory:flinkExample",new Properties())
   // prepare insert and update statements
   insertStmt = conn.prepareStatement("INSERT INTO Temperatures (sensor, temp) VALUES   (?, ?)")
   updateStmt = conn.prepareStatement("UPDATE Temperatures SET temp = ? WHERE sensor = ?")
}
override def invoke(r: SensorReading, context: Context[_]):Unit = {
   // set parameters for update statement and execute it
   updateStmt.setDouble(1, r.temperature)
   updateStmt.setString(2, r.id)
   updateStmt.execute()
   // execute insert statement if update statement did not update any row
   if (updateStmt.getUpdateCount == 0) {
       // set parameters for insert statement
       insertStmt.setString(1, r.id)
       insertStmt.setDouble(2, r.temperature)
       // execute insert statement
       insertStmt.execute()
  }
}
override def close(): Unit = {
   insertStmt.close()
   updateStmt.close()
   conn.close()
}
}

由于Apache Derby不提供内置的UPSERT语句，因此示例接收器首先尝试更新一行并插入新行(如果不存在具有给定键的行)，从而执行UPSERT写操作。当未启用WAL时，Cassandra sink连接器遵循相同的方法。

事务性sink连接器

当幂等接收器连接器不适合时，无论是应用程序输出的特性、所需接收器系统的属性，还是由于更严格的一致性要求，事务性接收器连接器都可以作为替代。如前所述，事务接收连接器需要与Flink的检查点机制集成，因为它们可能只在检查点成功完成时才向外部系统提交数据。

为了简化事务接收的实现，Flink的DataStream API提供了两个模板，可以扩展它们来实现自定义接收操作符。两个模板都实现了CheckpointListener接口来接收来自JobManager关于完成的检查点的通知(有关接口的详细信息，请参阅“接收关于完成的检查点的通知”):

• GenericWriteAheadSink模板收集每个检查点的所有输出记录，并将它们存储在sink任务的操作符状态。在失败的情况下，状态被检查并恢复。当任务收到检查点完成通知时，它将完成的检查点的记录写入外部系统。带有WAL- enabled的Cassandra sink连接器实现了这个接口。

• TwoPhaseCommitSinkFunction模板利用了外部接收器系统的事务特性。对于每个检查点，它启动一个新事务，并在当前事务的上下文中将所有后续记录写入sink系统。接收器在接收到相应检查点的完成通知时提交事务。

在下面，我们将描述接口及其一致性保证。

GENERICWRITEAHEADSINK

GenericWriteAheadSink通过改进一致性属性简化了sink操作符的实现。该操作符与Flink的检查点机制集成，目标是将每条记录精确地写入外部系统一次。但是，您应该知道存在这样的失败场景，即提前写日志接收器输出的记录不止一次。因此，一个GenericWriteAheadSink不能提供精确一次保证，只能提供至少一次的保证。我们将在本节后面更详细地讨论这些场景。

GenericWriteAheadSink的工作方式是将所有接收到的记录附加到一个写前日志中，这个日志由检查点分割。每次sink操作符接收到一个检查点barrier时，它都会启动一个新分片，并将所有后续记录附加到新分片中。WAL被存储并作为操作符状态进行检查。由于日志将被恢复，所以在失败的情况下不会丢失任何记录。

当GenericWriteAheadSink接收到关于完成的检查点的通知时，它会输出存储在对应于成功的检查点的segment中的所有记录。根据sink操作符的具体实现，可以将记录写入任何类型的存储或消息系统。当所有记录都成功输出后，必须在内部提交相应的检查点。

检查点通过两个步骤提交。首先，接收器持续存储提交的检查点信息，然后从WAL中删除记录。无法将提交信息存储在Flink的应用程序状态中，因为它不是持久性的，并且在出现故障时将被重置。相反，GenericWriteAheadSink依赖于一个名为CheckpointCommitter的可插入组件来存储和查找关于外部持久存储中提交的检查点的信息。例如，Cassandra sink连接器默认使用一个向Cassandra写入的CheckpointCommitter。

由于GenericWriteAheadSink的内置逻辑，实现一个利用WAL的sink并不困难。扩展GenericWriteAheadSink的操作符需要提供三个构造函数参数:

• 一个CheckpointCommitter，见前面章节介绍。

• 一个TypeSerializer用于序列化输入记录。

• 传递给CheckpointCommitter的作业ID，以标识跨应用程序重新启动的提交信息

此外，write-ahead运算符需要实现一个单一的方法:

boolean sendValues(Iterable<IN> values, long chkpntId, long timestamp)

GenericWriteAheadSink调用sendValues()方法将完成的检查点的记录写入外部存储系统。该方法接收一个Iterable(包括检查点的所有记录)、一个检查点的ID和一个生成检查点的时间戳。如果所有写操作都成功，则该方法必须返回true；如果写操作失败，则返回false。

示例8-14展示了一个写到标准输出的WriteAhead sink实现。它使用FileCheckpointCommitter，我们在这里不讨论它。您可以在包含该书示例的代码仓库中查找它的实现。

注意
GenericWriteAheadSink不实现SinkFunction接口。因此，不能使用DataStream.addSink()添加扩展GenericWriteAheadSink的sink，而是使用DataStream.transform()方法附加它。

val readings: DataStream[SensorReading] = ???
// write the sensor readings to the standard out via a writeahead log
readings.transform("WriteAheadSink", new SocketWriteAheadSink)
// -----
class StdOutWriteAheadSink extends GenericWriteAheadSink[SensorReading](
// CheckpointCommitter that commits checkpoints to the local filesystem
new FileCheckpointCommitter(System.getProperty("java.io.tmpdir")),
// Serializer for records
createTypeInformation[SensorReading].createSerializer(new ExecutionConfig),
// Random JobID used by the CheckpointCommitter
UUID.randomUUID.toString) {
override def sendValues(
   readings: Iterable[SensorReading],
   checkpointId: Long,
   timestamp: Long): Boolean = {
   for (r <- readings.asScala) {
  // write record to standard out
  println(r)
  }
  true
  }
}

示例代码库包含一个应用程序，该应用程序在发生故障时定期进行故障恢复，以演示StdOutWriteAheadSink和一个常规的DataStream.print() sink的行为。

如前所述，GenericWriteAheadSink不能提供精确一次保证。有两种失败情况会导致记录被输出不止一次:

• 当任务运行sendValues()方法时，程序失败。如果外部sink系统不能自动地写入多个记录(要么全部写入，要么没有写入)，那么可能已经写入了部分记录。由于检查点尚未提交，所以在恢复期间sink将再次写入所有记录。

• 所有记录都正确写入，sendValues()方法返回true;但是，在调用CheckpointCommitter或CheckpointCommitter未能提交检查点之前，程序会失败。在恢复期间，所有尚未提交的检查点记录将被重新写入。

TWOPHASECOMMITSINKFUNCTION

Flink提供了TwoPhaseCommitSinkFunction接口，以简化sink函数的实现，这些sink函数提供端到端的精确一次保证。但是，2PC sink函数是否提供这种保证取决于实现细节。我们从一个问题开始讨论这个接口:“2PC协议是不是太昂贵?”

通常，2PC是确保分布式系统一致性的昂贵方法。但是，在Flink上下文中，协议对于每个检查点只运行一次。此外，TwoPhaseCommitSinkFunction协议利用了Flink的常规检查点机制，因此增加的开销很小。TwoPhaseCommitSinkFunction的工作原理与WAL sink非常相似，但它不会收集Flink应用状态下的记录;相反，它将它们以开放事务的形式写入外部接收器系统。

TwoPhaseCommitSinkFunction实现以下协议。在sink任务发出第一个记录之前，它在外部sink系统上启动一个事务。所有随后收到的记录都是在事务的上下文中写入的。当JobManager启动一个检查点并在应用程序的源中注入barriers时，2PC协议的投票阶段就开始了。当操作符接收到barrier时，它会checkpoint状态，并在完成之后向JobManager发送确认消息。当sink任务接收到barrier时，它将持久化其状态，准备提交当前事务，并在JobManager上确认检查点。JobManager的确认消息类似于2PC协议的提交投票。sink任务必须尚未提交事务，因为不能保证作业的所有任务都将完成它们的检查点。sink任务还为在下一个检查点barrier之前到达的所有记录启动一个新事务。

当JobManager从所有任务实例接收到成功的检查点通知时，它将检查点完成通知发送给所有感兴趣的任务。此通知对应于2PC协议的提交命令。当接收任务接收到通知时，它提交以前检查点的所有打开的事务。sink任务一旦确认其检查点，就必须能够提交相应的事务，即使在出现故障的情况下也是如此。如果不能提交事务，接收器将丢失数据。当所有sink任务提交它们的事务时，2PC协议的迭代就成功了。

我们来总结一下外部sink系统的要求:

• 外部sink系统必须提供事务支持，或者sink必须能够模拟外部系统上的事务。因此，sink应该能够向sink系统写入数据，但是写入的数据在提交之前不能对外公开。

• 在检查点间隔期间，事务必须开启并接受写操作。

• 事务必须等到接收到检查点完成通知时，再提交。在恢复周期的情况下,这可能需要一些时间。如果sink系统关闭事务(例如,一个超时),未提交的数据将丢失。

• 处理一旦失败，sink必须能够恢复事务。一些sink系统提供一个事务ID可用于提交或中止一个开启的事务。

• 提交一个事务必须是一个幂等操作,sink或外部系统应该能够做到：一个事务已经提交或重复提交，没有影响。

通过一个具体的例子，可以更容易地理解sink系统的协议和需求。例8-15显示了一个TwoPhaseCommitSinkFunction，它只向文件系统写一次(精确一次)。实际上，这是前面讨论的BucketingFileSink的简化版本。

class TransactionalFileSink(val targetPath: String, valtempPath: String)
extends TwoPhaseCommitSinkFunction[(String, Double),String, Void](
createTypeInformation[String].createSerializer(new ExecutionConfig),
createTypeInformation[Void].createSerializer(new ExecutionConfig)) {
var transactionWriter: BufferedWriter = _
// Creates a temporary file for a transaction into which the records are written.
override def beginTransaction(): String = {
   // path of transaction file is built from current time and task index
   val timeNow = LocalDateTime.now(ZoneId.of("UTC")).format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
   val taskIdx = this.getRuntimeContext.getIndexOfThisSubtask
   val transactionFile = s"$timeNow-$taskIdx"
   // create transaction file and writer
   val tFilePath = Paths.get(s"$tempPath/$transactionFile")
   Files.createFile(tFilePath)
   this.transactionWriter = Files.newBufferedWriter(tFilePath)
   println(s"Creating Transaction File: $tFilePath")
   // name of transaction file is returned to later identify the transaction
   transactionFile
}
/** Write record into the current transaction file. */
override def invoke(
   transaction: String,
   value: (String, Double),
   context: Context[_]): Unit = {
   transactionWriter.write(value.toString)
   transactionWriter.write('
')
}
/** Flush and close the current transaction file. */
override def preCommit(transaction: String): Unit = {
   transactionWriter.flush()
   transactionWriter.close()
}
/** Commit a transaction by moving the precommitted transaction file
* to the target directory.
*/
override def commit(transaction: String): Unit = {
   val tFilePath = Paths.get(s"$tempPath/$transaction")
   // check if the file exists to ensure that the commit is idempotent
   if (Files.exists(tFilePath)) {
       val cFilePath = Paths.get(s"$targetPath/$transaction")
       Files.move(tFilePath, cFilePath)
  }
}
/** Aborts a transaction by deleting the transaction file. */
override def abort(transaction: String): Unit = {
val tFilePath = Paths.get(s"$tempPath/$transaction")
   if (Files.exists(tFilePath)) {
  Files.delete(tFilePath)
  }
}
}

TwoPhaseCommitSinkFunction[IN, TXN, CONTEXT]有三个类型参数:

• IN指定输入记录的类型。在例8-15中，这是一个带有String和Double的Tuple2。

• TXN定义了一个事务标识符，可用于在失败后识别和恢复事务。在例8-15中，这是一个包含事务文件名称的字符串。

• CONTEXT定义了一个可选的自定义上下文。例8-15中的TransactionalFileSink不需要上下文，因此将类型设置为Void。

TwoPhaseCommitSinkFunction的构造函数需要两个TypeSerializer----—一个用于TXN类型，另一个用于CONTEXT类型。

最后，TwoPhaseCommitSinkFunction定义了五个需要实现的功能:

• beginTransaction(): TXN启动一个新的事务并返回事务标识符。例8-15中的TransactionalFileSink创建一个新的事务文件，并将其名称作为标识符返回。

• invoke(txn: TXN, value: IN, context: Context[_]): Unit  将一个值写入当前事务。示例8-15中的sink将该值作为字符串追加到事务文件。

• preCommit(txn: TXN): Unit 预提交一个事务。预提交事务可能不会收到进一步的写操作。例8-15中的实现刷新并关闭事务文件。

• commit(txn: TXN): Unit 提交一个事务。此操作必须是幂等的—如果此方法被调用两次，不能将记录写入输出系统两次。在例8-15中，我们检查事务文件是否仍然存在，并将其移动到目标目录(如果存在的话)。

• abort(txn: TXN): Unit 中止一个事务。对于一个事务，此方法也可能被调用两次。例8-15中的TransactionalFileSink检查事务文件是否仍然存在，如果仍然存在，则删除它。

正如您所看到的，该接口的实现并不太复杂。然而，实现的复杂性和一致性保证取决于sink系统的特性和功能。例如，Flink的Kafka生成器实现了TwoPhaseCommitSinkFunction接口。如前所述，如果由于超时而回滚事务，连接器可能会丢失数据。因此，即使它实现了TwoPhaseCommitSinkFunction接口，也不能提供精确一次保证。

异步访问外部系统

除了接收或发送数据流之外，通过在远程数据库中查找信息来丰富数据流是另一个需要与外部存储系统交互的常见用例。一个例子就是著名的雅虎流处理基准测试，它基于需要用存储在键值存储中的相应活动的详细信息丰富广告点击流。

对于这些用例，最直接的方法是实现一个MapFunction，它为每个处理过的记录查询数据存储，等待查询返回结果，丰富记录，并输出结果。虽然这种方法很容易实现，但它存在一个主要问题:对外部数据存储的每个请求都增加了显著的延迟(一个请求/响应包含两条网络消息)，而MapFunction将大部分时间花在等待查询结果上。

Apache Flink提供AsyncFunction来减少远程I/O调用的延迟。AsyncFunction并发发送多个查询并异步处理它们的结果。可以将其配置为保留记录的顺序(请求返回的顺序可能与发送它们的顺序不同)，或者按照查询结果的顺序返回结果，以进一步减少延迟。该函数还与Flink的检查点机制进行了适当的集成——当前正在等待响应的输入记录是检查点的，在恢复的情况下，查询是重复的。此外，AsyncFunction可以正确地处理事件时间处理，因为它可以确保即使启用了无序结果，水印也不会被记录覆盖。

为了利用AsyncFunction，外部系统应该提供一个支持异步调用的客户端，这是许多系统的情况。如果系统只提供同步客户端，则可以创建线程来发送请求并处理它们。AsyncFunction的接口如下图所示:

trait AsyncFunction[IN, OUT] extends Function {
   def asyncInvoke(input: IN, resultFuture:ResultFuture[OUT]): Unit
}

函数的类型参数定义其输入和输出类型。使用两个参数为每个输入记录调用asyncInvoke()方法。第一个参数是输入记录，第二个参数是返回函数结果或异常的回调对象。在示例8-16中，我们展示了如何在DataStream上应用AsyncFunction。

val readings: DataStream[SensorReading] = ???
val sensorLocations: DataStream[(String, String)] =AsyncDataStream.orderedWait(
   readings,
   new DerbyAsyncFunction,
   5,
   TimeUnit.SECONDS, // timeout requests after 5 seconds
   100) // at most 100 concurrent requests

使用了AsyncFunction的异步操作符用AsyncDataStream对象配置，它提供了两个静态方法:orderedWait() 和unorderedWait()。这两个方法是重载方法的，使用不同的参数组合。orderedWait()应用一个异步操作符，它按照输入记录的顺序发出结果，而unorderWait()操作符只确保水印和检查点barrier保持对齐。其他参数指定记录的异步调用何时超时，以及启动多少并发请求。示例8-17显示了DerbyAsyncFunction，它通过JDBC接口查询嵌入式Derby数据库。

class DerbyAsyncFunction extends AsyncFunction[SensorReading, (String, String)] {
// caching execution context used to handle the query threads
private lazy val cachingPoolExecCtx = ExecutionContext.fromExecutor(Executors.newCachedThreadPool())
// direct execution context to forward result future to callback object
private lazy val directExecCtx =
ExecutionContext.fromExecutor(org.apache.flink.runtime.concurrent.Executors.directExecutor())
/**
* Executes JDBC query in a thread and handles the resulting Future
* with an asynchronous callback.
*/
override def asyncInvoke(
   reading: SensorReading,
   resultFuture: ResultFuture[(String, String)]): Unit = {
val sensor = reading.id
// get room from Derby table as Future
val room: Future[String] = Future {
// Creating a new connection and statement for each record.
// Note: This is NOT best practice!
// Connections and prepared statements should be cached.
val conn = DriverManager.getConnection("jdbc:derby:memory:flinkExample",new Properties())
val query = conn.createStatement()
// submit query and wait for result; this is a synchronous call
val result = query.executeQuery(s"SELECT room FROM SensorLocations WHERE sensor ='$sensor'")
// get room if there is one
val room = if (result.next()) {
result.getString(1)
} else {
"UNKNOWN ROOM"
}
// close resultset, statement, and connection
result.close()
query.close()
conn.close()
// return room
room
}(cachingPoolExecCtx)
// apply result handling callback on the room future
room.onComplete {
   case Success(r) => resultFuture.complete(Seq((sensor,r)))
   case Failure(e) => resultFuture.completeExceptionally(e)
}(directExecCtx)
}
}

示例8-17中的DerbyAsyncFunction的asyncInvoke()方法在Future中封装了阻塞JDBC查询，它是通过CachedThreadPool执行的。为了保持示例的简洁，我们为每个记录创建一个新的JDBC连接，当然，这是非常低效的，应该避免。Future[String]保存JDBC查询的结果。

最后，我们对Future应用一个onComplete()回调，并将结果(或可能的异常)传递给ResultFuture处理程序。与JDBC查询Future不同，onComplete()回调由DirectExecutor处理，因为将结果传递给ResultFuture是一个轻量级操作，不需要专门的线程。注意，所有操作都是以非阻塞方式完成的。

• 需要指出的是，AsyncFunction实例是按顺序调用其每个输入记录的——函数实例不是以多线程方式调用的。因此，asyncInvoke()方法应该通过启动异步请求并使用将结果转发到ResultFuture的回调来处理结果，从而快速返回。必须避免的常见反模式包括:

• 发送一个阻塞asyncInvoke()方法的请求。

• 发送异步请求，但在asyncInvoke()方法中等待请求完成。

结束语

在本章中，您将了解Flink DataStream应用程序如何从外部系统读取数据并将数据写入外部系统，以及应用程序实现不同端到端一致性保证的要求。我们介绍了Flink最常用的内置源和sink连接器，它们也代表不同类型的存储系统，如消息队列、文件系统和键值存储。

随后，我们向您展示了如何实现自定义源和sink连接器，包括WAL和2PC接收器连接器，并提供了详细的示例。最后，您了解了Flink的AsyncFunction，它可以通过异步执行和处理请求来显著提高与外部系统交互的性能。