Stream Processing with Apache Flink

1. 並列処理とプログラミングパラダイム

よく知られているように、計算集約的またはデータ集約的な作業、つまり比較的大きな計算量を必要とする処理には、並列計算や分割統治法が非常に有効な手段です。この手法の鍵となるのは、既存のタスクをどのように分割するか、あるいはコンピューティングリソースをどのように適切に割り当てるかという点です。

たとえば、学校の授業で教師が生徒にテスト答案の採点を依頼することがあります。テストに A、B、C の 3 問がある場合、生徒たちの分担と協力は次のような方法が考えられます。

方法 1：全テスト用紙の 3 問をそれぞれ異なる人に分担して採点する方法です。各採点者が担当する設問の採点を完了したら、次の採点者にテスト用紙を渡すことで、流れ作業のような効果が生まれます。ただし、設問は全部で 3 問しかないため、生徒の数が増えてもこの流れ作業の協業は拡大しにくくなります。

方法 2：方法 1 の分業をさらに拡張したものです。同じ設問を複数の生徒で分担できるようにします。たとえば、設問 A を 2 名、設問 B を 3 名、設問 C を 1 名で採点する場合です。この場合、計算タスクをさらに細かく分割する方法を考える必要があります。たとえば、全生徒を 3 つのグループに分け、第 1 グループが設問 A、第 2 グループが設問 B、第 3 グループが設問 C を担当します。第 1 グループの生徒たちは、グループ内でさらに作業を分割できます。たとえば、グループ A の 1 人目がテスト用紙の半分を採点し、2 人目が残りの半分を採点します。バッチ処理が完了したら、手元のテスト用紙を次のグループに渡します。
上述のように、テスト用紙の設問に基づく分割がいわゆる計算並列性であり、テスト用紙そのものの分割がデータ並列性です。

この並列性は上記の DAG で表現できます。

図の中で、設問 A を採点する生徒は追加のタスクを担っています。たとえば、テスト用紙を教員室から採点場所に運ぶ作業です。設問 C を担当する生徒にも追加のタスクがあり、全生徒の採点が完了した後に、合計得点の集計と提出記録の処理を行います。これに応じて、グラフ内のすべてのノードは 3 つのカテゴリに分類できます。第 1 のカテゴリは Source で、データを取得する役割（テスト用紙の受け取り）を担います。第 2 のカテゴリはデータ処理ノードで、ほとんどの場合外部システムとのやり取りは不要です。最後のカテゴリは、計算ロジック全体を外部システムに書き込む役割（得点の集計と提出記録）を担います。これら 3 種類のノードが、Source ノード、Transformation ノード、Sink ノードです。DAG において、ノードは計算を表し、ノード間の接続は計算間の依存関係を表します。

プログラミングについて

1 から 10 までの 10 個の数字を含むデータセットがあり、各数字を 2 倍にして累積和を求める場合（上図参照）、どのように実装すればよいでしょうか。方法はさまざまあります。

プログラミングで解決する場合、大きく 2 つのアプローチがあります。1 つ目は命令型プログラミングで、機械に対してデータ構造を段階的に生成する方法、それらのデータ構造を使って一時的な中間結果を保存する方法、およびそれらの中間結果を最終結果に変換する方法を一つずつ指示します。つまり、機械に手順を一つずつ指示するわけです。2 つ目は宣言型プログラミングです。宣言型では、通常、達成したいタスクだけを機械に伝え、命令型のような詳細な手順は指定しません。たとえば、元のデータセットを Stream に変換し、さらに Int 型の Stream に変換します。その過程で各数値を 2 倍にし、最後に sum メソッドを呼び出して全数値の合計を求めます。

宣言型プログラミング言語のコードはより簡潔で、この簡潔な開発スタイルこそが計算エンジンが追求する効果です。そのため、Flink のタスク記述関連の API はすべて宣言型です。

2. DataStream API の概要と簡単なアプリケーション

DataStream API の詳細な説明に入る前に、Flink API の論理レベルについて見ておきます。

旧バージョンの Flink では、API レベルは上図左側の 4 階層の関係に従っています。最上位は、より高度な API、つまりより宣言型の Table API や SQL を使ってロジックを記述できることを意味します。SQL や Table API で記述された内容は、内部的に Flink によって DataStream API のプログラムに変換・最適化されます。次の階層では、DataStream API のプログラムが一連の Transformation として表現され、最終的に Transformation は JobGraph（前述の DAG）に翻訳されます。

しかし、より新しいバージョンの Flink ではいくつかの変更が加えられており、主な変更は Table API と SQL 層に反映されています。DataStream API のプログラムに変換されるのではなく、直接底层の Transformation 層に到達するようになりました。つまり、DataStream API と Table API の関係が下位層から上位層へと変化したわけです。このプロセスの簡素化により、クエリ最適化のメリットがもたらされます。

次に、簡単な DataStream API プログラムを例に説明します。前述の 2 倍にして合計を求める要件です。

Flink で表現すると、基本的なコードは上図のようになります。スタンドアロンの例よりも少し複雑に見えるので、段階的に分解していきましょう。

まず、Flink で任意の機能を実装するには、対応する動作環境、すなわち Stream Execution Environment を取得する必要があります。

次に、環境を取得した後、環境の add Source メソッドを呼び出して、ロジックの初期データソース入力を追加します。データソースを設定した後、データソースの参照、つまり Data Source オブジェクトを取得できます。

最後に、一連の変換メソッドを呼び出して、Data Source 内のデータを変換できます。

この変換は図に示す通りで、各数値を 2 倍にし、その後、合計を求めるために keyBy を使ってデータをグループ化する必要があります。渡される定数は、全データを 1 つのグループにまとめることを意味し、最後にこのグループ内の全データを最初のフィールドに基づいて累積し、結果を取得します。結果を得た後、スタンドアロンプログラムのように単純に出力することはできず、ロジック全体に Sink ノードを追加して、全データをターゲットの場所に書き込む必要があります。上記の作業がすべて完了したら、Environment の Execute メソッドを呼び出して、上記で記述した全ロジックをリモートまたはローカルのクラスターに送信して実行する必要があります。

Flink の DataStream API プログラムとスタンドアロンプログラムの最大の違いは、プログラムの最初の数ステップではデータの計算がトリガーされないことです。DAG グラフを描くようなもので、論理 DAG グラフ全体が描画された後、Execute メソッドを通じてグラフ全体をまとめてクラスターに送信して実行します。

ここでの説明で Flink DataStream API と DAG グラフが結びつきます。実際の Flink タスクの生成プロセスは、上述の説明よりもはるかに複雑で、段階的な変換と最適化が必要です。下図は Flink ジョブの具体的な生成プロセスを示しています。

DataStream API で提供される変換操作

前述のサンプルコードに示すように、各 DataStream オブジェクトは対応するメソッドが呼び出されると新しい変換を生成します。それに対応して、底层では新しいオペレーターが生成され、既存の論理 DAG グラフに追加されます。これは、既存の DAG グラフの最終ノードを指す接続を追加することに相当します。これらの API はすべて呼び出し時に新しいオブジェクトを生成し、その新しいオブジェクトに対して引き続き変換メソッドを呼び出すことができます。この連鎖的な方法で DAG 図を段階的に描画していくわけです。

上記の説明には高階関数の考え方が含まれています。DataStream 上で変換を呼び出す際は、パラメーターを渡す必要があります。つまり、変換はデータに対してどのような操作を実行したいかを決定し、オペレーターに実際に渡される関数はその変換操作をどのように実行するかを決定します。

上図では、左側にリストされた API のほか、Flink DataStream API には非常に重要な 2 つの関数があります。ProcessFunction と CoProcessFunction です。これら 2 つの関数は、最下層の処理ロジックとしてユーザーに提供されます。上図左側の青い部分に含まれるすべての変換は、理論上、底层の ProcessFunction と CoProcessFunction で実装できます。

データパーティショニングについて

データパーティショニングとは、従来のバッチ処理におけるデータシャッフリングの操作を指します。トランプをデータに例えると、従来のバッチ処理のシャッフル操作は、カードをソートするプロセスに相当します。通常のカードゲームでは、カードを引きながら順番に並べ替え、同じ数字をまとめて配置します。この方法の最大のメリットは、カードを出す際に必要なカードを一度に見つけられることです。シャッフルは従来のバッチ処理のデータ再配置方法です。ストリーム処理ではデータが動的に流入するため、カードのソートやデータ処理、グループ化やパーティショニングのプロセスもリアルタイムで行われます。

たとえば、上図右側に示すように、上流のオペレーター A に 2 つの処理インスタンスがあり、下流のオペレーター B に 3 つの処理インスタンスがあるとします。ここで示されるシャッフルに相当するストリーム処理が、データパーティショニングまたはデータルーティングと呼ばれます。A がデータの処理を完了した後、下流の B のどの処理インスタンスに結果を送信するかを示すものです。

Flink で提供されるパーティショニング戦略

下図は Flink で提供されるパーティショニング戦略です。注意点として、DataStream が keyBy メソッドを呼び出すと、Key 値に基づいてデータ全体をパーティショニングできます。ただし厳密には、keyBy は実際には底层の物理パーティショニング戦略ではなく、変換操作です。API の観点から見ると、DataStream を KeyedDataStream に変換し、両者でサポートされる操作も異なります。

これらすべてのパーティショニング戦略の中で、Rescale は少し理解しにくいかもしれません。Rescale は上流と下流のデータの局所性に関与します。従来の Rebalance、つまりラウンドロビン（順番に割り当てる）と似ていますが、ネットワークを越えたデータ転送を回避しようとする点が異なります。

上記のすべてのパーティショニング戦略が適用できない場合は、PartitionCustom を呼び出してカスタムデータパーティショニングを定義できます。注意点として、これはカスタムユニキャストに過ぎません。つまり、各データに対して送信先の下流インスタンスを 1 つだけ指定でき、複数のコピーに複製して複数の下流インスタンスに送信する方法はありません。

Flink でサポートされるコネクタ

前述のように、下図には 2 つの重要なノードがあります。ノード A は外部システムに接続し、外部システムからデータを Flink 処理クラスタに読み込む必要があります。ノード C、すなわちシンクノードは、集約・処理した結果を何らかの外部システムに書き込む必要があります。ここでの外部システムは、ファイルシステムやデータベースなどです。

Flink の計算ロジックにはデータ出力がない場合もあります。つまり、最終データを外部システムに書き出さなくてもよいということです。Flink には状態の概念があり、中間で計算された結果は実際には State を通じて外部システムに公開できるため、専用の Sink を持たなくても構いません。ただし、すべての Flink アプリケーションには Source が必須です。つまり、どこかからデータを読み込まないと、後続の処理を実行できません。

Source と Sink のコネクタの種類について注意すべき点は次のとおりです。

Source については、データの更新を継続的に監視してアクセスし、対応する更新データをシステムに転送できるかどうかが重要な関心事となります。たとえば、Flink には CSV ファイルなどのファイルに対応する FileSystem コネクタがあります。CSV ファイルコネクタの定義時に、パラメーターを使って特定ディレクトリのファイル変更を継続的に監視し、更新されたファイルにアクセスするかどうかを指定できます。

Sink については、書き込み先の外部システムが書き込み結果の更新をサポートしているかどうかが重要な関心事となります。たとえば、Kafka にデータを書き込む場合、通常、データ書き込みは Append-Only です。つまり、システムに既に書き込まれたレコードを変更することはできません（コミュニティでは Kafka Compaction を使って Upsert Sink を実装しています）。データベースに書き込む場合は、主キーを使った既存データの更新が一般的にサポートされています。

上記の 2 つの特性が、Flink のコネクタが静的データを対象とするか動的データを対象とするかの重要なポイントを決定します。

補足として、上記のスクリーンショットは Flink 1.11 以降のものであり、Flink 1.11 でコネクタがリファクタリングされています。また、Table、SQL、API レベルのコネクタは、DataStream レベルのコネクタよりも多くのタスクを担当します。たとえば、述語やプロジェクション操作のプッシュダウンをサポートしているかどうかなどです。これらの機能はデータ処理の全体パフォーマンス向上に貢献します。

3. Flink における状態と時間

DataStream API を深く理解するには、状態と時間が把握すべき重要なポイントです。

すべての計算は、簡単にステートレスコンピューティングとステートフルコンピューティングに分けられます。ステートレスコンピューティングは比較的簡単です。ここに加算オペレーターがあると仮定すると、データのセットが入力されるたびに、すべて加算して結果を出力します。これは純粋関数に少し似ています。純粋関数とは、各計算結果が入力データにのみ依存し、以前の計算や外部状態が一切影響しないことを意味します。

ここでは主に Flink のステートフルコンピューティングについて説明します。枝拾いのミニゲームを例に挙げます。このゲームが優れていると思う点は、多くの状態を自ら記録していることです。たとえば、数日間オンラインでいなかった後、中の NPC と話すと、「長い間オンラインではありませんでしたね」と言われます。つまり、前回のオンライン時間を状態として記録し、NPC のセリフを生成する際にその状態の影響を受けるわけです。

このような状態付き計算を実現するには、以前の状態を記録し、その状態を新しい計算に注入する必要があります。具体的な実装方法は 2 つあります。

1 つ目は、状態データをオペレーターに入る前に抽出し、状態データと入力データを組み合わせてから、同時にオペレーターに入力して出力を得る方法です。この方法は Spark の StructureStreaming で採用されています。メリットは、既存のステートレスオペレーターを再利用できることです。

2 つ目は Flink の現在の方法で、オペレーター自体が状態を持ちます。オペレーターは新しいデータを受信して計算する際、新しい入力データの影響と既存状態の計算プロセスへの影響の両方を考慮し、最終的に結果を出力します。

計算エンジンも前述のゲームのようにますますインテリジェントになり、データ内の基本法則を自動的に学習し、適応的に計算ロジックを最適化して高い処理パフォーマンスを維持できるようになるべきです。

Flink のステートプリミティブ

Flink のステートプリミティブとは、コードを通じて Flink の状態を使用する方法を指します。基本的な考え方は、プログラミング時にネイティブ言語（Java や Scala など）が提供するデータコンテナを捨て、Flink のステートプリミティブに置き換えることです。

優れた状態サポートを持つシステムとして、Flink は内部にさまざまな種類のステートプリミティブを提供しています。大きく分けて、すべてのステートプリミティブは Keyed State と Operator State の 2 種類に分類できます。Operator State の適用場面は比較的少ないため、ここでは説明しません。Keyed State に焦点を当てて説明します。

Keyed State、すなわちパーティション化された状態です。パーティション化された状態の利点は、既存の状態をロジックが提供するパーティションに基づいて異なるブロックに分割できることです。パーティション内の計算と状態は結び付けられ、異なる Key 値間の計算と状態の読み書きは互いに分離されます。各キー値は、自身の計算ロジックと状態のみを管理すればよく、他のキー値に対応するロジックや状態を考慮する必要がありません。

Keyed State はさらに 5 つのカテゴリに細分できます。具体的には次のとおりです。

よく使われるもの：ValueState、ListState、MapState
あまり使われないもの：ReducingState と AggregationState
Keyed State は RichFunction 内でのみ使用できます。RichFunction と通常の従来の Function との最大の違いは、独自のライフサイクルを持っていることです。Keyed State の使用は次の 4 つのステップに分けられます。

第 1 ステップは、RichFunction 内で State をインスタンスの変数として宣言します。
第 2 ステップは、RichFunction の open メソッド内で、State の初期割り当て操作を行います。割り当て操作は 2 つのステップからなります。まず StateDescriptor を作成し、作成時に State に名前を指定します。次に RichFunction 内の getRuntimeContext().getState(...) を呼び出し、先ほど定義した StateDescriptor を渡すと、State を取得できます。
補足：ストリーミングアプリケーションが初めて実行される場合、取得される State は空です。途中段階から再起動された場合は、設定と以前に保存されたデータに基づいて復元されます。

第 3 ステップは、State オブジェクトを取得した後、RichFunction 内で対応する State の読み書きができます。ValueState の場合は、value メソッドを呼び出して対応する値を取得できます。Flink フレームワークがすべての状態への同時アクセスを制御・制限するため、ユーザーは並行性の問題を考慮する必要がありません。

Flink の時間

時間は Flink において非常に重要なポイントであり、State と補完関係にあります。一般的に、Flink エンジンでは 2 種類の時間が提供されています。1 つ目は処理時間で、2 つ目はイベント時間です。処理時間は現実世界の時間を表し、イベント時間はデータに含まれる時間です。データ生成過程でタイムスタンプなどのフィールドが付帯します。多くの場合、データに付帯するタイムスタンプを基準にして、時間単位でデータを処理する必要があります。

処理時間の処理は比較的シンプルです。順序不同などの問題を考慮する必要がないためです。イベント時間の処理は比較的複雑です。処理時間はシステム時刻を直接呼び出すため、マルチスレッドや分散システムの不確実性を考慮すると、各操作の結果が不定になる可能性があります。一方、イベント時間のタイムスタンプは各データに書き込まれているため、同じデータを複数回再生処理しても、付帯するタイムスタンプは変わりません。処理ロジックが変わらなければ、最終結果は比較的確定性が高くなります。

処理時間とイベント時間の違いです。

上図のデータを例に説明します。時間 1 から 7 まで順番に並んでいます。マシン時間の場合、各マシンの時刻は単調増加します。この場合、処理時間で取得する時間は、小さい順に並べられたデータに完璧に適合します。イベント時間の場合、遅延や配信の理由により、データの到着順と実際の生成順が異なる可能性があり、データがある程度順序不同になることがあります。この場合、データに付帯するタイムスタンプを活用して、データを粗粒度でグループ分けする必要があります。たとえば、データを 3 つのグループに分け、第 1 グループの最小時間が 1、第 2 グループの最小時間が 4、第 3 グループの最小時間が 7 とします。このグループ分けにより、グループ間でデータが昇順に並ぶことになります。

ある程度の順序不同を完全に解決して、システム全体がデータを基本的に順序通りに扱えるようにするにはどうすればよいでしょうか。1 つの解決策は、ウォーターマークと呼ばれるメタデータをデータの途中に挿入することです。上記の例で、最初の 3 つのデータが到着した後、3 以下のデータは来ないと仮定すると、ウォーターマーク 3 をデータストリーム全体に挿入できます。これで 3 以下のデータは来ないことを保証し、安心して独自の処理ロジックを実行できます。

まとめると、処理時間は使用する際に厳密に増加します。イベント時間にはある程度の順序不同があり、ウォーターマークで緩和する必要があります。

API の観点から、タイムスタンプの割り当てやウォーターマークの生成方法は比較的簡単で、2 つの方法があります。

1 つ目は、SourceFunction 内で内部的に提供される collectWithTimestamp メソッドを呼び出して、タイムスタンプ付きのデータを抽出する方法です。SourceFunction 内の emitWatermark メソッドを使ってウォーターマークを生成し、データストリームに挿入することもできます。

2 つ目は、SourceFunction 以外の場合、DataStream.assignTimestampsAndWatermarks メソッドを呼び出し、同時に 2 種類のウォーターマーク生成方法を渡す方法です。

1 つ目の種類は周期生成で、環境に値を設定します。たとえば、どのくらいの頻度（実時間）でシステムが自動的にウォーターマーク生成戦略を呼び出すかを指定します。

2 つ目の種類は特別なレコードに基づいて生成します。特別なデータに遭遇した場合、AssignWithPunctuatedWatermarks メソッドを使ってタイムスタンプとウォーターマークを割り当てられます。

補足：Flink には WatermarkAssigner などの一般的に使用される割り当て機能が組み込まれています。たとえば、固定データに対して、データに対応するタイムスタンプから固定時間を引いた値をウォーターマークとします。タイムスタンプ割り当てとウォーターマーク生成のインターフェイスについては、後続のバージョンで変更がある可能性があります。新しいバージョンの Flink では、上記の 2 種類の生成方法が統一されていることに注意してください。

時間関連の API

Flink でロジックを記述する際に使用する時間関連の API で、下図はイベント時間と処理時間に対応する API をまとめています。

アプリケーションロジックのインターフェイスサポートを通じて、3 つのことを実現できます。

1 つ目は、記録された時間の取得です。イベント時間は context.getTimestamp を使って調整するか、SQL オペレーターのデータフィールドから対応する時間を抽出できます。処理時間は currentProcessingTime を直接呼び出すことができ、内部ではシステム時刻を取得する静的メソッドを直接呼び出した値が返されます。

2 つ目は、ウォーターマークの取得です。実際、ウォーターマークの概念はイベント時間にのみ存在し、処理時間には存在しません。ただし処理時間において、何かをウォーターマークと見なさなければならない場合、それはデータ時間そのものです。つまり、timerService.currentProcessingTime メソッドを初めて呼び出した後に取得する値です。この値は現在の記録時間であるだけでなく、現在のウォーターマーク値でもあります。時間は常に前方に流れるため、この値を初めて呼び出した後、2 回目の呼び出しで最初の値より小さくなることは絶対にありません。

3 つ目は、タイマーの登録です。タイマーの役割はクリーンアップです。たとえば、将来のある時点でキャッシュをクリーンアップする必要がある場合などです。クリーンアップ作業は将来のある時点で実行されるべきなので、timerService の registerEventTimeTimer または registerProcessingTimeTimer メソッドを呼び出してタイマーを登録し、メソッド全体にタイマーコールバックの処理ロジックを追加します。対応するイベント時間または処理時間がタイマー設定時間を超えると、そのメソッドを呼び出して、自身で記述したタイマーのクリーンアップロジックを実行します。