Flink SQL導入記

 (hyperconnect.github.io)
3 ポイント 投稿者 GN⁺ 2025-02-25 | 1件のコメント | WhatsAppで共有
  • Hyperconnect Azar Matching Dev Teamは、CPUを96個使っていたモノリシックなFlinkレガシーアプリを分割するため、アプリケーションコードではなくSQLでストリーミング処理を実装する方式を選択した
  • 複数のFlink Appに分けると分離性は向上する一方で運用負荷が増えるため、チームのFlink内部への理解度を活かせるFlink SQLが、生産性と運用効率の面でより適していると判断した
  • Flink SQLはCheckpoint/Savepoint、JobManager HA、TaskManagerの再分配、ウィンドウ・結合・イベントタイム・ウォーターマーク、UDFとカスタムコネクタをサポートしており、ksqlDBやSpark Structured Streamingよりもチームの状況に合っていた
  • 運用環境はKubernetes上のSession mode Flink Clusterとして構成し、Flink SQL Gateway APIとGitHub Actionsを使ってGitOps方式でクエリ配備とJob停止を処理した
  • 約1年間安定して運用されながら拡大中だが、クエリ再配備とクラスタ設定変更は依然として不便であり、GitOps Controllerパターンベースの改善を計画している

重いレガシーストリーミングアプリを置き換えた背景

  • Azar Matching Dev Teamは複数のFlinkベースのアプリを管理しており、その中にCPUを96個使用する重いレガシーアプリがあった
  • このアプリは、複数のマッチイベント結合、条件付きイベント発行、Redisフラグ保存といった機能を一か所に集めたモノリシック構造だった
  • 全社インフラ作業で実行ノードを変更した後、アプリが正常に動作しなくなり、単純なチューニングだけでは素早く解決するのが難しかった
  • 重要なイベント結合機能は別プロジェクトの新しいFlinkアプリにすでに実装されていたため、イベント結合後の条件付きイベント発行とロジック実行部分を置き換える方式が必要だった

置き換え方式の比較

  • 1つのFlink Appとして実装すれば管理対象は少ないが、再び巨大なアプリになる可能性が高く、一部の障害が他の機能に影響するおそれがある
  • 複数のFlink Appに分ければ各アプリを独立して管理できるが、アプリ数が増えるほどクラスタ・リソース・配備の負担も大きくなる
  • Flink SQLはクエリでロジックを定義するため素早く開発でき、1つのクラスタだけを管理すればよいが、複雑なロジックをSQLだけで表現するのは難しく、クラスタ運用経験も必要になる
  • チームはFlink内部実装への理解が深まっている状態であり、Flink SQLが生産性と運用効率の面で利点があると判断した

Flink SQLを選んだ理由

  • Flink SQLではアプリケーションコードを直接書かなくても、SQLでイベントストリーミング処理アプリを実装できる
  • 高可用性(HA)の面で、Flinkは状態保持処理をサポートしており、CheckpointとSavepointによってJob状態を定期的または任意の時点で保存・復旧できる
    • JobManagerはleader-standby形式のHAモードで構成できる
    • TaskManagerの一部が失敗しても、Jobの再試行戦略に従って失敗したTaskManagerの処理を他のTaskManagerへ再分配できる
  • SQL構文だけでも主要なストリーミング処理機能を扱える
    • SELECTでデータ形式を変換し、WHEREでレコードをフィルタリングする
    • JOINで複数のストリームを結合し、UNIONでストリームをまとめられる
    • tumbling、hopping(sliding)、session windowのようなウィンドウ処理をサポートする
    • イベントタイム処理とウォーターマークによって遅延データの許容範囲を設定できる
  • UDFとカスタムコネクタによって標準機能外の要件にも拡張できる
    • 既存レガシーの大半はRedis SETまたはINCRコマンドを使うパターンであり、Flink公式のRedis Connectorがなかったため、Redis Connectorを自作して利用した
    • 当時はARRAY型の積集合を求めるビルトイン関数がなかったため、UDFとして実装してクエリで使用した

ksqlDB・Spark Structured Streamingとの比較

  • ksqlDBは社内でKafka向けに利用しているConfluentプラットフォームに含まれており、全社的な利用事例もあった
  • しかし、状態保持ストリーミング処理のHA動作では非効率だと判断した
    • 状態保持演算のフェイルオーバー時には、state変更履歴であるchangelogをすべてreplayする必要があり、フェイルオーバー時間が長くなる可能性がある
    • 処理ストリームのレプリカを持ち、changelogを内部stateに継続的に反映する方式では、レプリカ側でも同じ演算を行うため、リソース消費が2倍になる可能性がある
    • 関連内容はConfiguring ksqlDB for High Availability | Confluent Developerで確認できる
  • Spark Structured StreamingはSpark SQLエンジンベースのストリーミング処理エンジンである
    • 社内利用事例があり、UDFとCustom Sinkの作成も可能である
    • Flinkより規模が大きく、よく整備されたエコシステムを持つ
  • Sparkはマイクロバッチ単位で動作するため、レコード単位の遅延が発生しうえ、リアルタイム処理が重要な状況ではFlinkより不利になりうる
  • チーム内にSpark経験がほとんどなく、Custom Sinkの作成も必要だったため、Sparkをすぐに選ぶのは難しかった

クラスタ環境の構築

  • ローカルではFlink公式Webページからバイナリをダウンロードし、{FLINK_HOME}/bin/start-cluster.shでクラスタを起動できる
  • {FLINK_HOME}/bin/sql-client.shを実行するとFlink SQL CLIが開き、SELECT 1;のようなテストクエリを投入できる
  • クエリ投入後、Flink Web UIで投入したクエリがJobに変換されて実行されたことを確認できる
  • 2022年末にFlink SQL Gatewayがリリースされ、HTTPベースのクエリ投入が可能になった

Kubernetesベースの運用アーキテクチャ

  • 社内の大半のサービスがKubernetes上で動作しているため、Flink SQL ClusterもKubernetes上に構成した
  • 既存のFlink AppはすべてApplication modeで配備・運用されていた
    • 各アプリケーションごとに個別のクラスタを立ち上げる方式である
    • KubernetesではアプリごとにJobManager PodとTaskManager Podをそれぞれ起動して動作する
    • アプリ間の独立性と分離、作業ごとの設定や依存関係管理の面で有利だった
  • Flink SQLはすでに起動しているクラスタにJobを投入するため、JobManagerとTaskManagerをSession modeで起動する必要があった
  • クラスタはStand Alone Cluster on Kubernetesガイドをベースに構成した
  • HA環境はHigh-Availability with Standalone Kubernetes設定を参照し、high-availability.storageDirにはs3を使用した
  • Native Kubernetes方式は提供シェルスクリプトでクラスタを起動するため、deployment設定を直接定義して配備する社内インフラ環境には適していないと判断した

HAとS3連携設定

  • HAおよびS3連携のため、config.yamlに次の設定を使用した
high-availability.type: kubernetes
high-availability.storageDir: s3://{s3-path-for-flinksql-recovery}
kubernetes.cluster-id: {cluster-id}
kubernetes.namespace: {k8s-namespace}


# namespace 내의 service account 를 통해 Kubernetes cluster 에 접근할 수 있도록 권한을 부여하는 작업이 필요할 수 있습니다.
kubernetes.service-account: {k8s-service-account-for-flinksql}
  • HA環境ではJobManager podを2つ起動し、相互のアドレスが異なっていないとリーダー選出ロジックなどが正常に動作しない
  • JobManagerコンテナの実行引数は次のように設定した
args: ["start-foreground", "-D", "jobmanager.rpc.address=$(POD_IP)"]
  • この設定により、Kubernetes ConfigMapに現在リーダーとして選出されたJobManager pod情報と現在実行中のJob IDなどが保存され、HAに活用される

GitOps方式のクエリ配備

  • Flinkが直接提供するFlink SQL向けWeb UIや専用ツールはまだない
  • Hue連携事例をPoCとして検討したが、当時はFlink SQL Gatewayのバージョン互換性問題があり、追加開発が必要で、開発環境構築にも時間がかかった
  • 社内でGitOpsパターンを多く利用しているため、クエリを配備したりJobを停止したりするGitHub Actionsを実装した
  • Repository内にJobごとのフォルダを作り、実行するクエリをSQLファイルとしてまとめて置いた
  • GitHub Actionsはフォルダ名を受け取り、クエリを抽出するSQLファイルを特定する
  • 実装はFlink SQL Gateway REST APIを呼び出す方式で、シンプルでテストしやすいPythonで記述した

運用事例と障害対応

  • JobManagerが失敗した経験はなかったが、HA設定上はJobManagerが失敗すると別のJobManagerがリーダーに選出されて処理を継続できる
  • TaskManagerは時々失敗し、その大半はKubernetes QoSポリシーによりPodが再起動されるケースだった
  • 一部のTaskManagerが失敗しても、処理が他のTaskManagerに再分配されて継続することを確認した
  • クエリ失敗の大半は異常データ流入または計算資源不足が原因だった
    • JSONデータを読む際、誤ったJSONフォーマットはjson.ignore-parse-errorsオプションでエラーデータを無視できる
    • JSON_VALUEで特定pathのデータを抽出する際、値が存在しない、または型が異なることで発生するエラーはDEFAULT {VALUE} ON ERRORでデフォルト値を設定できる
    • TaskManager CPUが100%を超える、またはメモリ不足の場合は、TaskManagerリソースを増やすかクエリparallelismを増やしたうえで再配備する
  • クラスタ設定変更やUDF追加のためにクラスタを再起動すると、一部Jobが失敗することがあった
    • 原因はJob timeoutやretry設定が適切でない場合が多かった
    • Jobが早すぎる段階で再試行を打ち切らず、クラスタ再起動後に安定するまで再試行するようtimeoutとretry設定を修正した

クエリ変更とstate復元の制約

  • クエリ条件を修正して再配備する際、savepointでstateを復元できるのは、条件式の値変更のようなごく単純な修正に限られる
  • window条件が変わるとstateも変更され、互換性維持が難しくなり、savepointを使った復元が困難になることがある
  • stateを維持する必要があり、かつ要件が頻繁に変わるなら、アプリを直接実装する方式のほうが適している可能性がある

モニタリングポイント

  • Flinkには標準提供のmetricが多く、社内監視基盤と適切なMetric Reporterがあれば監視環境を容易に構成できる
  • numRunningJobsはクラスタで現在実行中のJob数を示し、値が急減したまま維持されていれば失敗したJobがあると判断できる
  • taskmanager.cpu.loadとtaskmanager.memory.usedでクラスタのリソース使用量を把握できる
  • busyTimeMsPerSecondでTaskManagerがどれだけ忙しいかをJobごとに確認できる
  • Kafkaをsourceとして使う場合、records-lag-maxでデータ遅延状況を素早く確認できる

例: Kafkaログインイベントのウィンドウ集計

  • Appendixの例では、Kafkaからイベントを受け取り、10秒ごとに直近1分間のログインイベント数をKafkaへ配信する
  • 入力データはJSONフォーマットで、event_time、event_type、data.user_idフィールドを含む
  • クエリではpipeline.name、parallelism.default、table.exec.state.ttlを設定する
  • 入力テーブルlogin_eventはKafka connectorとJSON formatを使い、json.ignore-parse-errorsをtrueに設定する
  • row_timeはevent_timeから生成し、ウォーターマークは現在までに観測されたイベントタイム基準で最大5秒遅れて到着するイベントを処理するよう設定する
  • 出力テーブルwindowed_login_countはKafka topicへ結果を配信し、proc_time AS PROCTIME()フィールドを含む
  • HOP(row_time, INTERVAL '10' SECOND, INTERVAL '1' MINUTE)で10秒間隔の1分hopping windowを作り、COUNT(*)でwindow内レコード数を集計する

運用結果と今後の改善点

  • チームは既存のFlink経験を活かし、以前よりも容易かつ迅速に複数の機能を追加できた
  • 生産性と運用効率の面で満足できる結果を得られたと評価している
  • 導入後約1年のあいだ、特別な運用作業なしで安定して動作した
  • 現在は段階的に運用拡大中である
  • クエリ再配備とクラスタ設定変更には不便さが残っており、GitOps Controllerパターン実装によるクエリ配備環境の改善を計画している

関連記事

1件のコメント

 
flgkselql98 2025-02-26

Flinkのような分散システムは、HAを維持するためにrackを2〜3個維持すべきですが、Kubernetesを連携することでHAを保証したように見えますね。ただ、結局はkube slave nodeのリソースも考えないといけないはずなので、Flinkだけを載せるnodeを構成したのか気になります（Flink高負荷時にslave nodeがダウンする問題がありそうです）。
そういう観点で、Kubernetesを使うメリットはあるのでしょうか？

また、Flinkでウィンドウ関数を使うと、その間のデータはメモリに保持されることでSQLのjoin文が動作するわけですが、trade-offの観点で見るとFlinkは良い選択肢なのかと考えてしまいます。時間が経つほど巨大化するSQL + jobが落ちたときに起きる大変なこと……。

私も最上流のdata sourceでjoinが必要な状況のとき、Flinkを使わずにどのような方式でapplication levelに下ろして処理できるか悩みますね。