外部数据访问的异步I / O.

外部数据访问的异步I / O.

本页介绍了Flink的API与外部数据存储的异步I / O的使用。对于不熟悉异步或事件驱动编程的用户，有关Futures和事件驱动编程的文章可能是有用的准备。

注：有关异步I / O实用程序的设计和实现的详细信息，请参阅提议和设计文档FLIP-12：异步I / O设计和实现。

需要异步I / O 算子操作

当与外部系统交互时（例如，当使用存储在数据库中的数据来丰富流事件时），需要注意与外部系统的通信延迟不会主导流应用程序的总工作。

天真地访问外部数据库中的数据，例如在a中MapFunction，通常意味着同步交互：将请求发送到数据库并MapFunction等待直到收到响应。在许多情况下，这种等待构成了函数的绝大部分时间。

与数据库的异步交互意味着单个并行函数实例可以同时处理许多请求并同时接收响应。这样，可以通过发送其他请求和接收响应来覆盖等待时间。至少，等待时间在多个请求上摊销。这导致大多数情况下流量吞吐量更高。

注意：通过仅扩展MapFunction到非常高的并行度来提高吞吐量在某些情况下也是可能的，但通常会产生非常高的资源成本：拥有更多并行MapFunction实例意味着更多任务，线程，Flink内部网络连接，网络连接到数据库，缓冲区和一般内部副本记录开销。

先决条件

如上一节所示，对数据库（或键/值存储）实现适当的异步I / O需要客户端访问支持异步请求的数据库。许多流行的数据库提供这样的客户端

在没有这样的客户端的情况下，可以通过创建多个客户端并使用线程池处理同步调用来尝试将同步客户端转变为有限的并发客户端。但是，这种方法通常比适当的异步客户端效率低。

异步I / O API

Flink的Async I / O API允许用户将异步请求客户端与数据流一起使用。API处理与数据流的集成，以及处理顺序，事件时间，容错等。

假设有一个目标数据库的异步客户端，则需要三个部分来实现对数据库的异步I / O流转换：

的实现AsyncFunction是把请求分派
一个回调，它接受算子操作的结果并将其交给ResultFuture
在DataStream上应用异步I / O 算子操作作为转换以下代码示例说明了基本模式：
Java
Scala

// This example implements the asynchronous request and callback with Futures that have the
// interface of Java 8's futures (which is the same one followed by Flink's Future)
/**
 * An implementation of the 'AsyncFunction' that sends requests and sets the callback.
 */
class AsyncDatabaseRequest extends RichAsyncFunction<String, Tuple2<String, String>> {
    /** The database specific client that can issue concurrent requests with callbacks */
    private transient DatabaseClient client;
    @Override
    public void open(Configuration parameters) throws Exception {
        client = new DatabaseClient(host, post, credentials);
    }
    @Override
    public void close() throws Exception {
        client.close();
    }
    @Override
    public void asyncInvoke(String key, final ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
        // issue the asynchronous request, receive a future for result
        final Future<String> result = client.query(key);
        // set the callback to be executed once the request by the client is complete
        // the callback simply forwards the result to the result future
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    return result.get();
                } catch (InterruptedException | ExecutionException e) {
                    // Normally handled explicitly.
                    return null;
                }
            }
        }).thenAccept( (String dbResult) -> {
            resultFuture.complete(Collections.singleton(new Tuple2<>(key, dbResult)));
        });
    }
}
// create the original stream
DataStream<String> stream = ...;
// apply the async I/O transformation
DataStream<Tuple2<String, String>> resultStream =
    AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);

/**
 * An implementation of the 'AsyncFunction' that sends requests and sets the callback.
 */
class AsyncDatabaseRequest extends AsyncFunction[String, (String, String)] {
    /** The database specific client that can issue concurrent requests with callbacks */
    lazy val client: DatabaseClient = new DatabaseClient(host, post, credentials)
    /** The context used for the future callbacks */
    implicit lazy val executor: ExecutionContext = ExecutionContext.fromExecutor(Executors.directExecutor())
    override def asyncInvoke(str: String, resultFuture: ResultFuture[(String, String)]): Unit = {
        // issue the asynchronous request, receive a future for the result
        val resultFutureRequested: Future[String] = client.query(str)
        // set the callback to be executed once the request by the client is complete
        // the callback simply forwards the result to the result future
        resultFutureRequested.onSuccess {
            case result: String => resultFuture.complete(Iterable((str, result)))
        }
    }
}
// create the original stream
val stream: DataStream[String] = ...
// apply the async I/O transformation
val resultStream: DataStream[(String, String)] =
    AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100)

重要提示：ResultFuture在第一次通话时完成ResultFuture.complete。所有后续complete调用都将被忽略。

以下两个参数控制异步算子操作：

超时：超时定义异步请求在被视为失败之前可能需要多长时间。此参数可防止死/失败的请求。
容量：此参数定义可能同时有多少异步请求正在进行中。尽管异步I / O方法通常会带来更好的吞吐量，但算子仍然可能成为流应用程序的瓶颈。限制并发请求的数量可确保算子不会累积不断增加的待处理请求积压，但一旦容量耗尽就会触发反压。

超时处理

当异步I / O请求超时时，默认情况下会引发异常并重新启动作业。如果要处理超时，可以覆盖该AsyncFunction#timeout方法。

结果顺序

由AsyncFunction一些未定义的顺序经常完成的并发请求，基于哪个请求首先完成。为了控制发出结果记录的顺序，Flink提供了两种模式：

无序：异步请求完成后立即发出结果记录。在异步I / O 算子之后，流中记录的顺序与之前不同。当使用处理时间作为基本时间特性时，此模式具有最低延迟和最低开销。使用AsyncDataStream.unorderedWait(…)此模式。
Ordered：在这种情况下，保存流顺序。结果记录的发出顺序与触发异步请求的顺序相同（算子输入记录的顺序）。为此，算子缓冲结果记录，直到其所有先前记录被发出（或超时）。这通常会在检查点中引入一些额外的延迟和一些开销，因为与无序模式相比，记录或结果在检查点状态下保持更长的时间。使用AsyncDataStream.orderedWait(…)此模式。

活动时间

当流应用程序与事件时间一起工作时，异步I / O算子将正确处理水印。这意味着两种订单模式具体如下：

无序：水印不会超过记录，反之亦然，这意味着水印建立了一个订单边界。记录仅在水印之间无序发出。只有在发出水印后才会发出某个水印后发生的记录。反过来，只有在水印发出之前输入的所有结果记录之后才会发出水印。

这意味着，在水印的存在，将无序的方式介绍了一些相同的延迟和管理开销的订购模式一样。开销量取决于水印频率。

Ordered：保存记录的水印顺序，就像保存记录之间的顺序一样。与处理时间相比，开销没有显着变化。

请记住，摄取时间是事件时间的一种特殊情况，其中自动生成的水印基于源处理时间。

容错保证

异步I / O 算子提供完全一次的容错保证。它将检查点中的飞行中异步请求的记录存储起来，并在从故障中恢复时恢复/重新触发请求。

实施技巧

对于实现期货有一个执行人（或执行上下文在Scala）回调，我们建议使用DirectExecutor，因为回调通常做最少的工作，并且DirectExecutor避免了额外的线程到线程切换的开销。回调通常只将结果传递给ResultFuture，将其添加到输出缓冲区。从那里开始，包括记录发射和与检查点副本记录交互的重要逻辑无论如何都发生在专用线程池中。

A DirectExecutor可以通过org.apache.flink.runtime.concurrent.Executors.directExecutor()或获得com.google.common.util.concurrent.MoreExecutors.directExecutor()。