このページは外部データストアを使った非同期I/OのためのFlinkのAPIの使用を説明します。非同期あるいはイベント駆動プログラミングに詳しくないユーザは、機能とイベント駆動プログラミングについての論文が事前に有用かも知れません。
注意: 非同期I/Oユーティリティの設計と実装についての詳細は、提案と設計のドキュメント FLIP-12: 非同期 I/O の設計と実装 で見つかるかもしれません。
外部システムと影響しあう場合(例えば、データベース内に格納されたデータを使ってストリームイベントを増やす場合)、外部システムとの通信遅延がストリーミングアプリケーションの全体の仕事を占めないように注意する必要があります。
例えばMapFunction
内で、外部システム内のデータへ無邪気にアクセスすることは、同期相互作用を意味します: リクエストはデータベースに送信され、MapFunction
は応答が受け取られるまで待ちます。多くの場合において、この待機は関数の時間の大多数を占めます。
データベースとの非同期な相互影響は、1の並行度の関数インスタンスが同時に多くのリクエストを処理することができ、応答を同時に受け取ることができることを意味します。つまり、待機時間は他のリクエストの送信と応答の受信と重なり合うことができます。少なくとも待機時間は複数のリクエストに償却されます。これはほとんどの場合において、かなり高いストリーミング スループットに繋がります。
注意: MapFunction
をとても高い並行度にスケーリングするだけのスループットの改善は、ある場合にはうまくいきますが、通常とても高いリソースの代償に付きます: より多くの平行な MapFunction インスタンスはより多くのタスク、スレッド、Flink内部ネットワーク接続、データベースへのネットワーク接続、バッファ および 一般的な内部予約の記帳を意味します。
上の章で説明されたように、データベースへの適切な非同期(あるいはキー/値 ストア)の実装は、非同期のリクエストをサポートするデータベースへのクライアントを必要とします。多くの人気のあるデータベースはそのようなクライアントを提供します。
そのようなクライアントが無い場合、複数のクライアントと生成し同期呼び出しをスレッドプールを使って処理することで、非同期クライアントを制限された同時クライアントに切り替えてみることができます。しかし、このやり方は通常適切な非同期クライアントよりも効率が悪いです。
Flinkの非同期 I/O API を使って、データストリームを持つ非同期リクエストクライアント使うことができます。API はデータストリームとの統合、処理順を調整、イベント時間、耐障害性などを処理します。
目的のデータベースのための非同期クライアントを持つと仮定すると、データベースに対する非同期I/Oを使ったストリーム返還を実装するために3つの部分が必要です:
AsyncFunction
の実装AsyncCollector
に手渡しするcallback以下のコード例は基本的なパターンを説明します:
// 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(final String str, final AsyncCollector<Tuple2<String, String>> asyncCollector) throws Exception {
// issue the asynchronous request, receive a future for result
Future<String> 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<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, 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)
以下の2つのパラメータは非同期オペレーションを制御します:
タイムアウト: タイムアウトは非同期リクエストが失敗したと見なされるまでどれだけ長く掛かるかを定義します。このパラメータはdead/failedリクエストを保護します。
容量: このパラメータはどれだけ多くの非同期リクエストが同時に進行中になることができるかを定義します。たとえ非同期I/Oのやり方が一般的により良いスループットに繋がるとしても、オペレータはまだストリーミングアプリケーションでのボトルネックになりえます。同時に発生するリクエストの数を制限することで、オペレータはこれまで成長している延期されたリクエストのバックログを収集しなくなるでしょう。しかしそれはいったん容量が使い尽くされるとバックプレッシャーを引き起こすでしょう。
AsyncFunction
によって発行される同時に発生するリクエストは、どのリクエストが最初に終了したかに基づいて、しばしば何らかの非定義の順番で完了します。結果のレコードが発行された順番に制御するために、Flinkは以下の2つのモードを提供します:
Unordered: 結果のレコードは非同期リクエストが完了するとすぐに発行されます。ストリーム内のレコードの順番は非同期I/Oの前と後とで異なります。基本的な時間の特性上、processing timeと一緒に使われた場合、このモードは最もレイテンシが低く、最もオーバーヘッドが低いです。このモードのためには AsyncDataStream.unorderedWait(...)
を使います。
Ordered: この場合、ストリームの順番は保持されます。結果のレコードは非同期リクエストが起動されたものと同じ順番(入力レコードのオペレータの順番)で発行されます。これを行うために、全ての処理中のレコードが発行(あるいはタイムアウト)されるまで、オペレータは結果のレコードをバッファします。レコードあるいは結果はチェックポイントされた状態の中でunorderedモードに比較して長期間維持されるため、これは通常余分なレイテンシとチェックポイント時の幾らかのおオーバーヘッドを導入します。このモードのためには、AsyncDataStream.orderedWait(...)
を使います。
ストリーミングアプリケーションがevent timeと連携する場合、ウォーターマークは非同期I/Oオペレータによって正しく処理されるでしょう。そのことは同時に2つの順番モードについて以下のことを意味します:
Unordered: ウォーターマークはレコードに追いつかず、逆も同様です。ウォーターマークが順番の境界を確立することを意味します。レコードは2つのウォーターマーク間でのみ順番無く発行されます。あるウォーターマークの後で発生するレコードは、ウォーターマークが発行された後でのみ発行されるでしょう。ウォーターマークが発行される前の入力からの全ての結果のレコードの後でのみ、今度はウォーターマークが発行されるでしょう。
ウォーターマークの前で、unordered モードがorderedと同じようになんらかの同じレイテンシと管理のオーバーヘッドを導入することを意味します。オーバーヘッドの量はウォーターマークの頻度に依存します。
Ordered: レコード間の順番が保持されるのと同じように、レコードのウォーターマークの順番が保持されます。processing timeとの連携に比べて、オーバーヘッドにはそれほど大きな変更はありません。
Ingestion Time は、ソースの処理時間に基づく自動的に生成されたウォーターマークを使ったevent time の特別な場合であることを思い出してください。
非同期I/Oオペレータは完全に確実に1回の耐障害性の保証を提供します。チェックポイント内の実行中の非同期リクエストのレコードを格納し、障害時の回復時にリクエストを回復/再起動します。
コールバックは一般的に最小の仕事を行い、DirectExecutor
は追加のスレッド-トゥ-スレッドの手渡しオーバーヘッドを避けるため、コールバックのためのExecutor (あるいはScalaでの ExecutionContext)を持つFutures を使った実装については、DirectExecutor
を使うことを提案します。コールバックは一般的に結果をAsyncCollector
に手渡しだけします。これは出力バッファに追加します。そこから、レコードの発行を含む重いロジックおよびチェックポイントの記帳とのやりとりがとにかく専用のスレッドプールで起こります。
DirectExecutor
はorg.apache.flink.runtime.concurrent.Executors.directExecutor()
あるいは com.google.common.util.concurrent.MoreExecutors.directExecutor()
を使って取得されるかもしれません。
AsyncFunction は Multi-Threaded とは呼ばれません。
ここで明示的に指摘したい良くある混同は、AsyncFunction
はマルチスレッドの形式で呼ばれないということです。たった一つのAsyncFunction
が存在し、それはストリームのそれぞれの部分の中の各レコードについて順次呼ばれます。asyncInvoke(...)
メソッドが早く返り、(クライアントによって)コールバックに頼らない限り、それは適切な非同期I/Oにはならないでしょう。
例えば、以下のパターンはasyncInvoke(...)
関数をブロックすることになり、従って非同期の挙動を無効にします:
結果が受け取られるまでlookup/queryメソッドがブロックの呼び出しをするデータベースクライアントの使用
asyncInvoke(...)
メソッド内の非同期クライアントによって返されるfuture-typeオブジェクト上のブロック/待機