+
+{% highlight java %}
+// 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> {
+
+ /** 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(final String str, final AsyncCollector> asyncCollector) throws Exception {
+
+ // issue the asynchronous request, receive a future for result
+ Future resultFuture = 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 collector
+ resultFuture.thenAccept( (String result) -> {
+
+ asyncCollector.collect(Collections.singleton(new Tuple2<>(str, result)));
+
+ });
+ }
+}
+
+// create the original stream
+DataStream stream = ...;
+
+// apply the async I/O transformation
+DataStream> resultStream =
+ AsyncDataStream.unorderedWait(stream, new AsyncDatabaseRequest(), 1000, TimeUnit.MILLISECONDS, 100);
+
+{% endhighlight %}
+
+
+{% highlight scala %}
+/**
+ * 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, asyncCollector: AsyncCollector[(String, String)]): Unit = {
+
+ // issue the asynchronous request, receive a future for the result
+ val resultFuture: 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 collector
+ resultFuture.onSuccess {
+ case result: String => asyncCollector.collect(Collections.singleton((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)
+
+{% endhighlight %}
+
+
+
+*Note:* Improving throughput by just scaling the `MapFunction` to a very high parallelism is in some cases possible as well, but usually
+comes at a very high resource cost: Having many more parallel MapFunction instances means more tasks, threads, Flink-internal network
+connections, network connections to the database, buffers, and general internal bookkeeping overhead.
+
+
+## Prerequisites
+
+As illustrated in the section above, implementing proper asynchronous I/O to a database (or key/value store) requires a client
+to that database that supports asynchronous requests. Many popular databases offer such a client.
+
+In the absence of such a client, one can try and turn a synchronous client into a limited concurrent client by creating
+multiple clients and handling the synchronous calls with a thread pool. However, this approach is usually less
+efficient than a proper asynchronous client.
+
+
+## Async I/O API
+
+Flink's Async I/O API allows users to use asynchronous request clients with data streams. The API handles the integration with
+data streams, well as handling order, event time, fault tolerance, etc.
+
+Assuming one has an asynchronous client for the target database, three parts are needed to implement a stream transformation
+with asynchronous I/O against the database:
+
+ - An implementation of `AsyncFunction` that dispatches the requests
+ - A *callback* that takes the result of the operation and hands it to the `AsyncCollector`
+ - Applying the async I/O operation on a DataStream as a transformation
+
+The following code example illustrates the basic pattern:
+
+