10 traps to know before using Flink

新しいフレームワークの採用には、いつも多くの驚きが伴います。サービスが異常動作している原因を数日間かけて調査した結果、ある関数の使い方が間違っていたり、シンプルな設定が不足していたりしただけだったということは少なくありません。

Contentsquare[1] では、ますます増え続けるデータの厳しい要件に対応するため、データ処理タスクを継続的にアップグレードする必要があります。そこで、セッション[2]処理用の毎時 Spark タスクを Flink[3] ストリーミングサービスに移行することにしました。これにより、Flink のより堅牢な処理能力を活用し、ユーザーによりリアルタイムなデータと既存データを提供できます。しかし、それは簡単なことではなく、チームは 1 年間にわたって取り組みを続けてきました。同時に、驚くべき問題にもいくつか遭遇しました。この記事では、そのような落とし穴を回避するためのお手伝いをします。

1. 並列度設定による負荷の偏り

まず簡単な質問から始めましょう。Flink UI でジョブのサブタスクを調査していると、各サブタスクのデータ処理量に関して次のような奇妙な状況に遭遇することがあります。

各サブタスクのワークロードが均等ではありません

これは、各サブタスクのオペレータが受け取る Key Group の数が均等でないことを示しています。Key Group は、すべての可能なキーの一部を表します。あるオペレータが 1 つの Key Group を受け取り、別のオペレータが 2 つを受け取る場合、2 番目のサブタスクは約 2 倍の作業をこなす必要があります。Flink のコードを見ると、次の関数が見つかります。

この関数の目的は、すべての Key Group を実際のオペレータに分配することです。Key Group の総数は maxParallelism パラメータで決まり、オペレータの数は並列度と同じです。ここでの最大の問題は maxParallelism のデフォルト値で、特に指定がない場合、operatorParallelism + (operatorParallelism / 2) となります[4]。並列度を 10 に設定すると、maxParallelism は 15 になります（実際の最大同時実行数の下限は 128、上限は 32768 ですが、ここでは便宜上の数値です）。このように、上記の関数に基づいて、どのオペレータにどの Key Group が割り当てられるかを計算できます。

デフォルト設定では、一部のオペレータには 2 つの Key Group が割り当てられ、一部のオペレータには 1 つの Key Group しか割り当てられません

この問題の解決は非常に簡単です。同時実行数を設定する際に、maxParallelism にも値を設定し、その値を並列度の倍数にします。これにより、負荷がより均等になり、将来のスケールアウトもしやすくなります。

2. mapWithState と TTL の重要性に注意する

無限に多くのキーを含むデータを処理する場合、キー付きステートの保持ポリシー（TTL タイマーにより、所定の時間後に未使用データをクリーンアップする）を考慮することが重要です。ここで「無限」という用語は少し誤解を招く可能性があります。128 ビットでエンコードされたキーを扱う場合、キーの最大数には上限があります（2 の 128 乗に等しい）。しかし、それは途方もなく大きな数です。それほど多くの値をステートに保存できることはまずないため、キースペースは事実上無限であり、新しいキーは時間の経過とともに出現し続けると考えるのが最善です。

キー付きステートが Flink のデフォルトウィンドウのいずれかに含まれている場合は安全です。TTL を使用しなくても、ウィンドウの要素を処理する際に明示的なタイマーが登録され、clearAllState 関数が呼び出され、そのウィンドウに関連するステートとそのメタデータが削除されます。

Keyed State Descriptor[5] を使用してステートを管理する場合は、TTL 設定を簡単に追加でき、ステート内のキー数が増え続けるのを防げます。

ただし、よりシンプルな mapWithState メソッドを使いたくなるかもしれません。このメソッドは valueState へのアクセスを提供し、操作の複雑さを隠蔽します。テストや少数のキーを持つデータには問題ありませんが、本番環境で無限に増えるキーを扱う場合は問題を引き起こす可能性があります。ステートが隠蔽されているため TTL を設定できず、デフォルトでは TTL が設定されません。そのため、RichMapFunction のような宣言を行うなど追加の作業を検討する価値があります。これにより、ステートのライフサイクルをより細かく制御できます。

3. チェックポイントからの復元とパーティション再編

大規模なステートを使用する場合、増分チェックポイント（incremental checkpointing）が不可欠です。このケースでは、タスクのフルステートは約 8 TB で、15 分ごとにチェックポイントを取得するよう設定していました。チェックポイントは増分方式のため、15 分ごとに約 100 GB のデータをオブジェクトストレージに送信するだけで済み、フルチェックポイントよりもはるかに高速でネットワーク負荷も軽減されます。これはフォールトトレランスの面で非常にうまく機能しますが、タスク更新時にステートを復元する必要もあります。一般的なアプローチは、実行中のジョブに対してセーブポイントを作成することです。セーブポイントにはポータブルな形式でステート全体が含まれます。

しかし、このケースではセーブポイントの完了に数時間かかり、リリースのたびに長く煩雑なプロセスとなっていました。そこで、保持済みチェックポイント（Retained Checkpoints）[6] を使用することにしました。このパラメータを設定することで、時間のかかるセーブポイントをトリガーすることなく、以前のジョブのチェックポイントからステートを復元してリリースを高速化できます。

また、セーブポイントの方がチェックポイントより移植性は高いものの、保持済みチェックポイントでもジョブのパーティション変更が可能です（すべての種類のジョブで機能するわけではないため、事前にテストすることをお勧めします）。これはセーブポイントからのパーティション再編とまったく同じですが、Flink が TaskManager 間でデータを再配布する長時間のプロセスを伴いません。これを試したところ、完了に約 8 時間かかり、持続可能な状態ではありませんでした。幸い、RocksDB ステートバックエンドを使用しているため、このステップでスレッド数を増やして高速化できます。具体的には、以下の 2 つのパラメータを 1 から 8 に増やします。

予約済みチェックポイントを使用し、RocksDB 転送に割り当てるスレッド数を増やすことで、リリースとパーティション再編の時間を 10 分の 1 に短縮できました。

4. 事前にログを増やしておく

これは当たり前のように思えるかもしれませんが、忘れてしまいがちな点でもあります。ジョブを開発する際は、それが長期間実行され、予期しないデータを処理する可能性があることを念頭に置いてください。そのような場合、問題の調査に必要な情報をできるだけ多く確保しておき、同じデータを再度処理して問題を再現する必要をなくすことが重要です。

このタスクは、イベントを収集し、特定のルールに従って結合するものです。これらのルールの一部は通常は問題なく動作しますが、データスキューが発生すると非常に時間がかかります。あるタスクが 3 時間も停滞しているのを発見し、何をしているのか全く分かりませんでした。1 つの TaskManager の CPU だけが正常に稼働していたため、特定のデータが原因でアルゴリズムのパフォーマンスが低下しているのではないかと疑いました。

最終的にデータ処理が完了するとすべて正常に戻りましたが、どこから確認すべきか分かりませんでした。そこで、このようなケースに備えて予防的なログを追加しました。ウィンドウの処理中に所要時間を計測し、ウィンドウの計算に 1 分以上かかった場合は、可能な限りすべてのデータを記録します。これは、パフォーマンス低下を引き起こしているスキューの原因を正確に理解するのに非常に役立ちました。次回同じ現象が発生したときには、マージ処理が遅い理由の一部を特定できました。重複データを受信した場合、実際には数時間かかることがあります。ただし、過度なログ出力はパフォーマンスを低下させる可能性があるため注意が必要です。したがって、例外時にのみ情報が表示される適切な閾値を見つけることが重要です。

5. 停滞したジョブが実際に何をしているかを特定する方法

上記の問題の調査により、ジョブが停滞していると思われる場合に、現在実行中のコードを簡単に特定する方法が必要であることに気づきました。幸い、これを実現する簡単な方法があります。まず、TaskManager の JMX を設定してリモートモニタリングを有効にする必要があります。Kubernetes デプロイメントでは、3 つのステップで JMX に接続できます。

まず、flink-conf.yaml にこのプロパティを追加します
次に、ローカルポート 1099 を TaskManager の Pod のポートに転送します
最後に、jconsole を起動します

これにより、JVM 上の対象 TaskManager の情報を簡単に表示できます。停滞したジョブについては、稼働中の唯一の TaskManager を特定し、実行中のスレッドをプロファイリングしました。

JConsole により、各スレッドが現在何をしているかを確認できます

さらに詳しく調べると、1 つのスレッドを除くすべてのスレッドが待機状態であることが分かります（上記のスクリーンショットで強調表示）。これにより、ジョブがどのメソッド呼び出しで停滞しているかを素早く特定し、簡単に修正できました。

6. あるステートから別のステートへのデータ移行のリスク

実際の状況に応じて、異なるセマンティクスを持つ 2 つの異なるステート記述子を管理する必要があるかもしれません。たとえば、WindowContent ステートを通じて進行中のセッションのイベントを蓄積し、処理済みセッションを HistoricalSessions という ValueState に移動します。2 番目のステートは、後で必要になる場合に備えて数日間保持され、TTL が切れると破棄されます。

最初のテストはうまく動作しました。処理済みセッションに追加データを送信すると、同じキーに対して新しいウィンドウが作成されます。ウィンドウの処理中に、HistoricalSessions ステートからデータを取得して新しいデータを古いセッションとマージし、結果として得られるセッションは履歴セッションの拡張版となり、期待通りの結果でした。

この処理中にメモリ関連の問題が数回発生しました。いくつかのテストを経て、OOM が発生するのは古いデータを Flink に送信した場合のみ、つまり現在のウォーターマークより前のタイムスタンプを持つデータを送信した場合のみであることが分かりました。これにより、現在の処理方法に大きな問題があることが発覚しました。古いデータを受信すると、Flink は古いウィンドウとマージします。その間、古いウィンドウのデータはまだ WindowContent ステートに残っています（これは AllowedLateness の設定で実現されます）。そして結果のウィンドウは HistoricalSessions の内容とマージされますが、ここにも古いデータが含まれています。最終的に得られるのは重複イベントであり、同じセッションでいくつかのイベントが受信されると、各イベントが数千の重複を持つことになり、OOM に繋がります。

この問題の解決策は非常にシンプルです。WindowContent の内容を第 2 のステートに移動する前に、自動的にクリアされるようにします。これには Flink の PurgingTrigger を使用します。ウィンドウが満了したときに、ステートの内容をクリアするメッセージを送信します。

7. ReducingState と ProcessWindowFunction の比較

前述の通り、Flink の使用は、特定のキーのデータを蓄積し、そのすべてのデータをマージすることに依存しています。これには 2 つの方法があります。

ListState コンテナにデータを保存し、セッションが終了するのを待って、セッション終了時にすべてのデータをマージする

新しいイベントが到着するたびに ReducingState を使用して、前のデータとマージする

1 番目と 2 番目のどちらの方法を使うかは、WindowedStream で実行する関数によって決まります。ProcessWindowFunction を使用する process 呼び出しは ListState を使用し、ReduceFunction を使用する reduce 呼び出しは ReducingState を使用します。

ReducingState の利点は明白です。ウィンドウ処理前にすべてのデータを保存するのではなく、継続的に集約して 1 つのレコードにまとめます。これにより、通常はステートサイズが小さくなります。ただし、reduce 操作中にどれだけのデータが破棄されるかによります。このケースでは、HistoricalSessions に保存する 7 日間のデータに比べるとステートサイズは無視できるほど小さく、ストレージ面での改善はほとんどありませんでした。その代わりに、ListState を使用することでパフォーマンスの向上が確認できました。

その理由は、連続する reduce 操作では、新しいイベントが到着するたびにデータを逆シリアル化およびシリアル化する必要があるためです。これは RocksDBReducingState[7] の add 関数で確認でき、getInternal[8] を呼び出すことでデータの逆シリアル化が発生します。

一方、RocksDB で ListState の値を更新する場合、シリアル化は発生しません[9]。これは RocksDB のマージ操作のおかげで、Flink は逆シリアル化なしにデータを追加できます。

最終的に、ストレージへの影響を最小限に抑えながらレイテンシの削減に役立つパフォーマンスの向上を理由に、ListState を選択しました。

8. 入力データを信頼しない

入力が期待通りであると決して想定しないでください。タスクが偏ったデータ、重複データ、予期しないスパイク、無効なレコードを受信するなど、さまざまな未知の状況が発生する可能性があります。常に最悪のケースを想定し、タスクをこれらの問題から保護してください。

後で使用する重要な用語をいくつか定義しておきます。

「ページビュー (PV) イベント」は、主に受信するメッセージです。ビジターがクライアント側で URL を読み込む際に、userId、sessionNumber、pageNumber などの情報とともにトリガーされます。

「セッション」は、ユーザーがサイトを離れることなく行うすべてのインタラクションの合計を表します。Flink によって PV イベントやその他の情報を集約して計算されます。

タスクを保護するために、できるだけ多くの事前フィルタリングを追加しました。遵守すべきルールは、ストリームのできるだけ早い段階で無効なデータをフィルタリングし、中後段での不要な高コスト操作を避けることです。たとえば、特定のセッションに対して 300 を超える PV イベントを送信できないというルールがあります。各 PV イベントには、セッション内での位置を示すページ番号がインクリメントされて付けられます。セッション内で 300 を超える PV イベントを受信した場合、次の方法でフィルタリングできます。

特定のウィンドウが満了したときに PV イベントの数をカウントする

ページ番号が 300 より大きいイベントを破棄する

1 番目のオプションはページ番号の値に依存しないためより信頼性が高いように見えますが、除外する前に 300 件以上の PV イベントをステートに蓄積する必要があります。最終的に 2 番目のオプションを選択しました。これにより、Flink に入力される時点で不良データを排除できます。

これらのステートレスフィルターに加えて、各キーに関連付けられたメトリクスに基づいてデータを除外する必要もあります。たとえば、セッションあたりの最大バイト数は 4 MB に設定されています。この数値はビジネス上の理由と、Flink における RocksDB のステートに関する制約に対処するために選ばれました。実際、Flink が使用する RocksDB API の値が 2^31 バイト[10]を超えると、処理は失敗します。したがって、前述の通り ListState を使用する場合は、データを蓄積しすぎないようにする必要があります。

新しく消費されたイベントの情報だけでは、セッションの現在のサイズを知ることは不可能です。つまり、ページ番号の場合と同じ手法は使えません。そこで、各キー（各セッション）のメタデータを RocksDB 内の別の ValueState に保存します。このメタデータは、keyBy オペレータの後、ウィンドウ処理の前に使用・更新されます。これにより、このメタデータに基づいて特定のキーの値の受け入れを停止するタイミングを把握できるため、RocksDB が ListState にデータを蓄積しすぎるのを防げます。

9. イベントタイミングの危険性

イベントタイム処理は大半のケースで優れていますが、遅延到着データの処理方法が時間のかかるものである場合、いくつかの望ましくない結果が生じる可能性があることを念頭に置く必要があります。この問題は Flink に直接関連するものではありません。外部コンポーネントが Kafka トピックにデータを書き込んでいる間に Flink がそのトピックからデータを消費している場合、その外部コンポーネントに問題があるとデータが遅延して到着します。具体的には、そのコンポーネントが一部のパーティションを他のパーティションよりも遅く消費する場合です。

このコンポーネント（Asimov と呼ばれます）は、Kafka トピックを読み取り、JSON データを解析し、protobuf に変換して別の Kafka トピックにプッシュするシンプルな Akka ストリーミングプログラムで、Flink が protobuf を処理できるようにします。Asimov の入力は各パーティション内で順序が保たれているはずですが、入力パーティションと出力トピックが一対一でマッピングされていないため、Flink が最終的にメッセージを処理する際に順序の乱れが発生する可能性があります。Flink はウォーターマークを遅らせることで順序の乱れに対応できるため、これ自体は問題ありません。

問題は、Asimov が一部のパーティションを他よりも遅く読み取った場合に発生します。Flink のウォーターマークは最も速い Asimov の入力パーティションに合わせて進みます（Flink の入力パーティションはすべて正常に進んでいるためです）。遅いパーティションからは古いタイムスタンプのレコードが送信されます。その結果、Flink はこれらのレコードを遅延データとして扱います。これ自体は問題ない場合もありますが、このジョブでは遅延レコードを処理するために特定のロジックを使用しており、RocksDB からデータを取得し、追加のメッセージを生成して下流の更新を実行する必要があります。つまり、Asimov が何らかの理由でいくつかのパーティションの処理に遅れるたびに、Flink はより多くの作業をこなす必要があります。

128 パーティションのトピックで、8 つのパーティションだけが遅延を蓄積し、Flink でのデータ到着が遅延しました

この問題に対して 2 つの回避策を見つけました。

Asimov の入力トピックを出力トピックと同じ方法（userId ベース）でパーティション分割できます。これにより、Asimov がいくつかのパーティションで遅延した場合、Flink 入力の対応するパーティションも遅延し、ウォーターマークの進行が遅くなります。

この方法は採用しませんでした。Asimov の前段で遅延到着データがある場合、同じ問題が存在し続けるため、各トピックを同じ方法でパーティション分割する必要が生じるからです。しかし、これは多くのケースで実現できません。

もう 1 つの解決策は、遅延イベントのバッチ処理に依存します。遅延イベントの処理を遅延させることができれば、イベントごとに 1 つの更新ではなく、セッションあたり最大 1 つの更新を保証できます。

カスタムトリガーを使用して、遅延イベント到着時にウィンドウを即座にトリガーしないようにすることで、2 番目の解決策を実装できます。デフォルトの EventTimeTrigger の実装で確認できる通り、遅延イベントは特定の条件下でタイマーを登録しません。このシナリオでは、とにかくタイマーを登録し、ウィンドウは即座に満了しません。ビジネス要件がこの方法でのバッチ更新を許容しているため、上流で遅延が発生した場合に数百もの高コストな更新を生成しないことを保証できます。

10. Flink にすべてを保存しない

最後に一般的なポイントを述べて終わりにします。データが大きく、あまり頻繁にアクセスされない場合は、Flink の外部に保存する方が良いでしょう。ジョブを設計する際は、必要なすべてのデータが Flink ノード上で直接利用可能であること（RocksDB またはインメモリ）が望ましいです。確かに、この種のデータを扱うにははるかに高速ですが、データが大きい場合はジョブに大きなコストが加わります。これは、Flink のステートはレプリケートされないため、ノードを失うとチェックポイントからの完全な復元が必要になるためです。定期的に数百 GB のデータをチェックポイントストレージに書き込む必要がある場合、チェックポイント処理自体も高コストになります。

ステートへのアクセスがパフォーマンス要件上重要な要素である場合は、Flink に保存する価値が確かにあります。ただし、追加のレイテンシを許容できるなら、レプリケーション機能を備え、特定のレコードへの高速アクセスが可能な外部データベースに保存する方が、多くの問題を回避できます。このユースケースでは、WindowContent ステートを RocksDB に保持しましたが、HistoricalSessions データは Aerospike[11] に移行しました。これにより、Flink ジョブがより高速になり、ステートが小さくなるため保守も容易になりました。さらに、Flink に残るデータが十分に小さくメモリに収まるため、RocksDB やローカル SSD を使用する必要がなくなるという利点も得られました。