The practice of Apache Flink in mobile cloud real-time computing

1. リアルタイムコンピューティングプラットフォームの概要

クラウド上のリアルタイムコンピューティングエンジンの進化は、いくつかの段階に分かれます。

2015 年から 2016 年にかけて、第 1 世代のリアルタイムコンピューティングエンジンである Apache Storm を使用していました。

2017 年、独自開発フレームワークと統合でき、運用保守の負荷とメンテナンスコストを軽減できる Apache Spark Streaming の調査を開始しました。

2018 年、ユーザーのクラウドコンピューティングへのニーズがますます増加し、Storm と Spark ではビジネス要件を十分に満たせなくなりました。同時に、ストリームコンピューティングに関する複数の著名な論文を調査したところ、Apache Flink が論文で言及されているセマンティクスを比較的完全に提供していることを発見しました。

2019 年から 2020 年にかけて、クラウドサービスの導入を開始し、リアルタイムコンピューティングプラットフォームをパブリッククラウドおよびプライベートクラウド向けにリリースしました。

2020 年から 2021 年にかけて、リアルタイムデータウェアハウスの調査を開始し、クラウド上で LakeHouse をオンライン化しました。

現在、Flink は主に China Mobile のシグナリングデジタル処理、リアルタイムユーザープロファイル、ログ収集、リアルタイムデータウェアハウス、リアルタイム運用保守モニタリング、リアルタイムレコメンデーション、クラウドデータパイプラインサービスに使用されています。


China Mobile のリアルタイムコンピューティングプラットフォームの機能は、3 つのパートに分かれます。

第 1 のパートはサービス管理で、タスクライフサイクル、Flink および SQL ジョブ、Spark Streaming ジョブのホスティングをサポートし、複数バージョンのエンジンに対応しています。

第 2 のパートは SQL サポートで、オンライン Notebook での記述、SQL 構文検出、UDF 管理、メタデータ管理を提供します。

第 3 のパートはタスク運用保守で、リアルタイムタスクのログ検索、リアルタイムパフォーマンス指標の収集、メッセージ遅延アラート、タスクバックプレッシャーアラートなどをサポートしています。

日常的なタスクシナリオで、ユーザーのプログラムデバッグコストが比較的高く、ユーザーが新バージョンのエンジンを試す期間も比較的長いことが判明しました。

複数バージョンの提出プロセスは次のとおりです。ユーザーのタスクはまず rtp サービスに提出され、rtp サービスはユーザープログラムを HDFS にアップロードして保存し、提出時に HDFS から取得して YARN クラスターに提出します。このようなタスクには共通点があります。Apache Flink のコアパッケージがジョブに含まれており、多くの問題を引き起こします。


そこでまず、ビジネス側と連携してジョブパッケージに Flink のコアパッケージを含めないようにしました。しかしメリットは比較的小さかったため、プラットフォーム側でテストを実施し、ユーザーが jar パッケージをアップロードする際にユーザーパッケージにコアパッケージが含まれているかどうかを積極的に検出するようにしました。違法なコアパッケージを含むジョブが見つかった場合、ユーザーの提出をブロックします。

このようなシンプルな操作により、会社に大きなメリットをもたらしました。

第 1 に、低価値バグの特定コストを大幅に削減しました。
第 2 に、ジョブのアップグレードとロールバックがより便利になりました。
第 3 に、運用の安定性と安全性が向上しました。

日常的なビジネスシナリオでは、ログ検索を通じてプロセスの複雑なロジックを検証する必要があります。さらに、ネイティブ TM の UI ログは開けず、フリーズしやすく、TM UI は検索をサポートしていません。前述の図のように、ビジネスロジックが非常に複雑な場合、Flink UI は上記の機能を提供できません。そこで、リアルタイムタスクログ検索機能を設計しました。


リアルタイムタスクログ検索の設計では、以下の課題を検討する必要があります。異なるマシン上に分散する TM のジョブプログラムログをどのように収集するか。ジョブに侵入せずにログを収集するにはどうすればよいか。ジョブが無駄なログを大量に出力しないように制限するにはどうすればよいか。

第 1 の課題には、プッシュモードを使用してログ収集の負荷を軽減します。
第 2 の課題には、Spring の AOP メカニズムを参照し、AspectJWeaver を使用します。エントリーポイントは Log4j の入力またはイベントで、その後ログを Sender に送信します。
第 3 の課題には、RateLimiter を使用してレート制限を実施します。

前述の図はリアルタイムタスクログ検索の全体設計です。ネイティブ TaskManager の下に AOP レイヤーを追加し、ログはまず TaskManager を通じてタスクに送信され、その後 AOP に送信されます。アスペクトアプローチにより、AOP 全体はユーザーに対して透過的です。その後 RateLimiter に送信され、さらに Sender に送信されます。RateLimiter はレート制限操作を実行します。その後ログは Kafka に継続的に送信され、検索時に Elasticsearch に送信されます。

リアルタイムタスクログ検索により、ビジネスプログラムは修正なしでログ検索をサポートできます。同時に、開発者はビジネスロジックを簡単に検証できます。スロットリング対策のおかげで、ログストレージのボトルネックは発生しません。さらに、プラットフォーム管理の負荷も軽減されます。

2. ビジネス最適化

このビジネスの出現は、観光部門、緊急部門、交通業界など、各レベルの政府部門のユーザーリソースデータに対するニーズを解決することです。たとえば、観光地などの重点エリアにおける交通計画、交通調査、人口フローモニタリング、流動人口のモニタリング管理などです。

モバイルユーザーの高カバレッジ率に依存し、通信ネットワークの地域サービス技術と GIS 技術を利用して、ユーザーシグナリングデータの統計を通じて、都市人口や移動性などの要素を分析・予測し、都市計画、交通計画、管理、リソース割り当て、移民人口管理、政策策定などの政府管理行動に意思決定データサポートを提供します。

1 日あたりの平均ビジネスデータは約 10 PB、1 日あたり 20 兆件で、単一データサイズは 0.5 KB です。インターネットデータ 2345 G、位置シグナリング、省市、ネットワークタイプ、インターフェースタイプなどを含みます。データ処理もより複雑で、データ暗号化、圧縮、バージョン統一などがあります。前述の図はシグナリング番号処理時の条件とビジネスロジックを示しています。

要件を簡素化し、クラスターに応答します。これはレポートゲートウェイです。各地からシグナリングデータをアップロードし、Flume クラスターからデータを受信して Hadoop クラスターに転送します。前述の図からわかるように、Flume と Hadoop の間には物理的なファイアウォールがあります。


データ量の増加に伴い、多くの問題にも遭遇しました。

第 1 に、Flume クラスターが常にアラートを発し、Flume チャネルフルを通知します。
第 2 に、ファイアウォールが制限を超えるとアラートが発せられます。
第 3 に、Flume が Kafka に書き込む際、Kafka sender がタイムアウトアラートを発します。
第 4 に、ダウンストリームでシグナリングデータを処理する際、Spark Streaming の処理が不安定です。

上記の問題は 2 つのカテゴリに要約できます。

第 1 のカテゴリは書き込みパフォーマンスの問題です。Kafka は書き込み時に頻繁にタイムアウトし、プロデュースパフォーマンスにボトルネックがあります。また、Flume はデータ送信時に NIC の上限速度に到達できません。

第 2 のカテゴリはアーキテクチャ設計の問題です。アーキテクチャに関与する多くのコンポーネントにより、メンテナンスコストが高くなっています。さらに、コンポーネントの責任が明確ではなく、Flume 内のデータクレンジングロジック、Spark ロジックと処理ロジックが複雑で、複数のシャッフルがあり、処理パフォーマンスが不安定です。


まず解決すべきは PRO が Kafka に書き込む際のタイムアウト問題です。この問題を解決するために、以下の最適化を実施しました。

ファイアウォールポートを最適化しました。

Kafka サーバー側で一部のパフォーマンスパラメーターを調整しました。

Kafka サーバー側で一部のパフォーマンスパラメーターのチューニングを実施しました。

しかし、Flume が Kafka に書き込む際のタイムアウト問題は完全に解決できなかったため、クライアントに焦点を当てました。第 1 は、特に batch.size、buffer.memory、request.time.out のチューニングにおいて、クライアントのパラメーターをどのように最適化するかです。第 2 は、単一マシン上で何並列のクライアントを設定するかという、単一マシンの NIC 最大速度を達成する方法です。

実践により、batch.size が 256 メガバイトで buffer.memory が 128 メガバイトの場合、パフォーマンスが最適になることが判明しました。しかしこの時点では NIC の最大速度にはまだ到達していません。

そこで第 2 ラウンドのテストを実施し、compression.type を追加して、送信データを圧縮することで送信帯域幅を増加させることを期待しましたが、結果は期待に沿いませんでした。

これは、低バージョンの Kafka に問題があるためです。検証スクリプト内のパラメーターの各値が同じであるため、圧縮率が比較的大きくなります。しかし実際の本番環境では各数値が異なるため、圧縮率は非常に小さくなります。

もう 1 つの課題は、NIC の最大速度を達成する方法です。最も簡単な方法は並列度を上げることですが、並列度が大きいほど良いわけではありません。実践により、同時実行数が 4 の時に NIC の最大速度に到達できることが判明しました。4 を超えると平均消費時間が大幅に増加し、Kafka 書き込みタイムアウトも引き起こします。

第 2 のポイントは Flume チャネルフルの問題です。

サービスを拡張する際、サービスのトランザクション API 処理は比較的ローレベルで、手動で処理する必要があります。さらに、サービスのトランザクションデータを処理する際にデータをコピーする必要があります。前述の図のように、ソースからチャネルにデータが送信される際、まずメモリーにデータのコピーが作成され、チャネルからシンクに送信される際にチャネルからメモリーに再度コピーされます。このプロセスの 2 回のコピーがリソースを浪費します。一方、Flink はトランザクション処理時にステート管理に依存するため、処理パフォーマンスが比較的安定しています。さらに、Flink は豊富なソースとシンクを持ち、スケーラビリティが比較的強力です。

そこで、Flume の代わりに Flink を使用して問題を解決することを決定しました。Flink に置き換えた後、収集パフォーマンスが向上し、大量データ転送のパフォーマンスボトルネックが解決され、安定性が大幅に向上しました。同時にコンポーネントの責任が明確化され、元のサービスに存在したすべてのロジックをバックエンドのリアルタイムデータ分解に移行し、収集層はデータ集約に集中し、処理層はデータソートに集中できるようにしました。さらに、技術スタックを統一し、Flink フレームワークをエンドツーエンドで採用することで、より高いパフォーマンスを実現し、開発と運用保守のコストを削減しました。

その結果、全体のパフォーマンスが 3 分の 1 向上し、メンテナンスコストも削減されました。

3. 安定性プラクティス

ジョブの安定性は主にサービス障害とソリューションを指します。サービス障害には主に、ジョブ障害、ジョブ消費遅延、ジョブ OOM、ジョブ再起動が含まれます。対応するソリューションは、ジョブの物理的隔離、サービスの縮退、リソース監視の強化、サービスの分割です。

プラットフォームメンテナーが最も懸念しているのは全体の問題です。

ZooKeeper クラスターのサーバーでネットワークサービス中断が発生すると、大量のタスク再起動も引き起こされます。Flink JobManager は ZooKeeper を使用してリーダーの選出と CheckpointID のカウンター管理を行います。

そこで ZooKeeper ネットワーク状態の遷移を分析しました。クライアントが ZooKeeper クラスターに接続する際、まず connected 状態です。ネットワーク切断後、Suspended 状態に変化します。Suspended 状態は lost 状態に変化し、その後 reconnected 状態に継続的に変化します。Flink は ZooKeeper 使用時に Curator 2.0 コンポーネントに依存しています。しかしこのコンポーネントには欠陥があり、Suspended 状態に遭遇するとリーダーを直接破棄します。これにより大部分のジョブが再起動し、ビジネスでは受け入れられません。

4. 将来方向の探求

今後は主に以下の 2 つの方向で探求を続けます。

第 1 はリソース利用率の方向です。Elastic Scaling の調査と K8s YuniKorn リソースキューの調査を含みます。Flink のクラウド移行後にリソースキュー問題があることを発見したため、キューごとにユーザーのリソースを管理する必要があります。
第 2 はデータレイクの方向です。第 1 は統一ストリームバッチサービスゲートウェイです。リアルタイムデータウェアハウスの構築時には Flink や Spark など異なるエンジンが使用される可能性があります。これらは 2 つの異なるサービスセットに属するため、統一ストリームバッチサービスゲートウェイが必要です。次にデータリネージ、データアセット、Data Quality サービスです。