Bytedance Flink status query practice and optimization

1. 背景

ご存知の通り、Flink の State は演算子の計算プロセスの中間結果を保存します。タスクに異常が発生した際、タスクスナップショット内の State をクエリすることで有効な手がかりを得られます。

しかし、Flink SQL タスクではジョブの状態をクエリしたい場合、State の定義方法や具体的な型を把握できないため、クエリのコストが非常に高くなりがちです。

この課題を解決するため、ByteDance のストリーミングコンピューティングチームは Flink SQL 上での State Query ソリューションを社内で提案しました。ユーザーは SQL を記述するだけで簡単に State をクエリできます。本記事では、ByteDance による Flink 状態クエリの関連取り組みを主に紹介します。

2. State Processor API の概要

状態クエリといえば、Flink 1.9 で導入された機能である State Processor API を連想します。State Processor API を使用すると、ジョブが生成したセーブポイントを DataSet に変換し、DataSet API を使って状態のクエリ、変更、初期化などの操作を完了できます。

以下は State Processor API を使った状態クエリの簡単な手順紹介です。

* まず、ExistingSavepoint を作成してセーブポイントを表現します。ExistingSavepoint の初期化にはセーブポイントのパスや StateBackend などの情報を指定します。

* 次に、ReaderFunction を実装して必要な状態を再登録し、状態の処理方法を定義します。状態クエリのプロセスでは、すべてのキーがトラバースされ、定義した方法に従って状態が処理されます。

* 最後に、Savepoint.readKeyedState を呼び出して演算子の UID と ReaderFunction を渡すことで、状態クエリを完了できます。

次に、状態クエリの背後にある原理を簡単に説明します。セーブポイントのディレクトリには 2 種類のファイルが含まれます。1 つは上図の opA-1-state のような状態データファイルで、最初のサブタスクにおける演算子 A の詳細データが含まれています。もう 1 つはメタデータファイルで、各演算子と状態ファイルのマッピング関係が保存されています。

状態クエリの実行時、まずクライアント側でセーブポイントのパスに基づいてメタデータファイルが解析されます。演算子 ID を通じて、クエリ対象の状態に対応するファイルのハンドルを取得できます。実際の状態クエリ実行時には、状態の読み取りを担当するタスクが新しい StateBackend を作成し、状態ファイルのデータを StateBackend に復元します。状態の復元が完了すると、すべてのキーがトラバースされ、対応する状態が ReaderFunction に渡されて処理されます。

ここで、既に状態クエリの機能が提供されているのに、なぜ独自に同じ作業を行うのかと疑問に思う方もいるかもしれません。これは主に State Processor API の使用中にいくつかの課題が発見されたためです。

* 状態をクエリするたびに個別の Flink バッチタスクを開発する必要があり、ユーザーにとって一定の開発コストがかかります。

* ReaderFunction の実装には、状態の名前、型、State Descriptor など、タスク状態の定義方法を明確に把握する必要があり、ユーザーにとって参入障壁が高くなっています。

* State Processor API では単一の演算子の状態しかクエリできず、複数の演算子の状態を同時にクエリできません。

* タスクが使用している状態や状態の型など、タスク状態のメタ情報を直接クエリできません。

総じて、2 つの目標があります。1 つはユーザーの利用コストを削減すること、もう 1 つは状態クエリ機能を強化することです。ユーザーが最もシンプルな方法で State をクエリできるようにし、同時に詳細な情報を把握しなくても済むようにしたいと考えています。

さらに、複数の演算子の状態を同時にクエリできたり、ジョブが使用している状態や各状態の型を直接クエリできるようにすることも望んでいます。

そこで、Flink SQL 上での State Query ソリューションを提案します。簡単に言えば、State をデータベースのように扱い、ユーザーは SQL を記述するだけで簡単に State をクエリできるというものです。

このソリューションでは 2 つの課題を解決する必要があります。

状態の詳細情報をユーザーから隠す方法：State Processor API を参照すると、状態のクエリにはセーブポイントのパス、StateBackend の型、演算子 ID、State Descriptor など多くの情報を提供する必要があることが分かります。これらの複雑な情報を SQL 文で完全に表現するのは明らかに困難です。では、状態のクエリには具体的に何が必要で、どのようにしてユーザーから状態内部の複雑な詳細を隠せるのか。これが最初の課題です。

状態を SQL で表現する方法：Flink における状態のストレージ方法はデータベースとは異なります。状態のクエリプロセスをどのように SQL で表現できるか。これが解決すべきもう 1 つの課題です。

3. StateMeta スナップショットの仕組み

まず最初の疑問に答えましょう。ある State をクエリするにはどのような情報が必要でしょうか。

前述の State Processor API の例を参照すると、ExistingSavepoint や ReaderFunction の作成時に、セーブポイントのパス、バックエンドの型、OperatorID、演算子キーの型、状態名、シリアライザなどの情報を提供する必要があります。これらを総称して状態のメタ情報と呼ぶことができます。

Flink SQL タスクにおいて、ユーザーがこれらの情報を明確に把握するのは非常に困難です。そこで、ユーザーはセーブポイント ID という最もシンプルな情報だけを提供すればよく、その他のメタ情報は Flink フレームワークがセーブポイントに保存することで、ユーザーから State の複雑な詳細を隠し、状態クエリを完了できるようにするという方針を採用しました。このため、StateMeta スナップショットの仕組みを導入しました。

StateMeta スナップショットとは、簡単に言えば状態のメタ情報をセーブポイントのメタデータに追加するプロセスです。具体的な手順は以下の通りです。

* まず、状態の登録時にタスクは演算子名、ID、キーシリアライザ、StateDescriptors などのメタ情報をタスクのメモリに保存します。

* セーブポイントがトリガーされると、タスクは状態のスナップショットと同時にメタ情報のスナップショットも取得します。スナップショットの完了後、状態のメタ情報と状態ファイルハンドルを JobManager に報告します。

* すべてのタスクから提出された StateMeta 情報を受信した後、JobManager はこれらのメタ情報を統合し、統合されたメタ情報をセーブポイントディレクトリ内の stateInfo という名前のファイルに保存します。

その後、状態クエリ時にはセーブポイント内の stateInfo ファイルを解析するだけで済み、ユーザーがコードでこれらのメタ情報を入力する必要はなくなります。これにより、ユーザーの状態クエリコストを大幅に削減できます。

4. State をデータベースとして表現する

次に 2 つ目の疑問に答えましょう。どのように SQL で State を表現するかです。実は、State Processor API の設計時にコミュニティでも State As Database という解決策が提案されていました。

従来のデータベースでは、テーブルは通常 Catalog、Database、Table の 3 つの要素で表現されます。同じロジックを Flink の State にも適用できます。Flink の State を特別なデータソースと見なし、ジョブが生成する各セーブポイントを独立した DB とみなします。この DB 内で状態のメタ情報と状態の詳細データを異なるテーブルとして抽象化し、ユーザーに公開します。ユーザーはこれらのテーブルを直接クエリしてタスクの状態情報を取得できます。

まず、State をテーブルとして表現する方法を見ていきましょう。Flink では一般的に KeyedState と OperatorState の 2 種類の State が使用されます。

OperatorState は Value という 1 つの属性のみを持ち、この State の具体的な値を表現します。したがって、OperatorState は Value フィールドのみを含むテーブル構造として表現できます。

KeyedState については、異なる Key と名前空間の下で各 State の値が異なる可能性があるため、Key、Namespace、Value の 3 つのフィールドを含むテーブル構造として表現できます。

単一の State を抽象化した後、複数の State を表現するのは比較的簡単です。上図の例では、この演算子に 2 つの KeyedState と 1 つの OperatorState の合計 3 つの状態が含まれています。これらのテーブルを UNION で結合し、state_name フィールドで異なる状態を区別するだけで、この演算子内のすべての状態を表現できます。

最後に、あるタスクがどの状態を使用しているか、またはそれらの状態の具体的な型をどう把握するかという疑問が残ります。

この問題を解決するため、StateMeta という特別なテーブルを定義しました。これは Flink タスク内のすべての State のメタ情報を表現するものです。StateMeta には各 State の名前、State の演算子 ID、演算子名、Key の型、Value の型などが含まれており、ユーザーは StateMeta テーブルを直接クエリすることでタスク内のすべての状態のメタ情報を取得できます。

5. Flink Batch SQL を使用したタスク状態のクエリ

以上が状態クエリスキームの全体的な紹介です。では実際にどのように State をクエリするのでしょうか。Word Count タスクを例に説明します。

まず、Flink SQL タスクを作成して起動します。web-ui で確認すると、このタスクには Source、Aggregate、Sink の 3 つの演算子が含まれています。次にセーブポイントをトリガーし、正常に作成されたら対応するセーブポイント ID を取得します。このセーブポイント ID を使用してジョブの状態をクエリできます。

Flink SQL タスクでの状態の使用状況が分からない場合、まずこの Flink タスクに含まれる状態とその型をクエリする必要があります。この情報は StateMeta テーブルから取得できます。上図のシナリオ 1 に示すように、StateMeta テーブルをクエリすると、このタスクに ListState と ValueState が含まれており、それぞれ Source 演算子と Aggregate 演算子に存在していることが分かります。

また、Flink に詳しい方ならご存知のように、Kafka ソースの State は現在のコンシュームオフセット情報を記録するために使用されます。シナリオ 2 に示すように、Source 演算子の状態をクエリすることで、タスクでコンシュームしている ApsaraMQ for Kafka のトピックのパーティションとオフセット情報を取得できます。

もう 1 つの一般的なシナリオとして、下流のビジネス担当者が特定のキー (key_662 など) の結果に異常を発見した場合のデバッグ方法があります。問題の切り分け時にジョブ内の Aggregate 演算子の状態を直接クエリし、同時にキーを key_662 に指定してクエリ条件とします。上図のシナリオ 3 に示すように、クエリ結果から key が 662 のときに対応する集約結果が 11290 であることが確認できます。このようにして、ユーザーは状態が正しいかどうかを簡単に検証できます。

6. 今後の展望

今後、State の機能をさらに充実させる予定です。現在は SQL を使用した State クエリ機能をサポートしていますが、コミュニティでは状態の変更や初期化の機能も提供されています。いくつかのシナリオではこれらの機能も重要です。たとえば、状態内の特定のキーの計算結果が誤っていることが分かっており、この部分のデータを修正したい場合や、タスクロジックの変更後に以前の状態と完全に互換でなくなり、状態を変更・初期化して新しいセーブポイントを生成したい場合があります。利用方法についても、SQL の INSERT 構文や UPDATE 構文を直接使用して状態の変更と初期化を完了できるようにすることを計画しています。

さらに、State の使いやすさをさらに向上させます。DAG 編集ソリューションを使用してジョブのトポロジー変更時の状態非互換性問題を解決していますが、Flink SQL タスクでフィールドを変更すると状態シリアライザが変更され、状態の非互換性が生じる可能性があります。この状況に対応するため、完全な Flink SQL ステートスキーマ進化ソリューションを設計しており、変更後の Flink SQL タスクの状態復元能力を大幅に向上させることができます。このソリューションは現在実装中です。さらに、タスクのオンライン化前に状態の互換性をチェックしてユーザーに通知する包括的な事前チェック機能も提供し、状態の非互換性によるジョブ起動失敗の影響を本番環境で回避できるようにする予定です。