Ideal Auto の K8s 上の Flink に基づくデータ統合の実践

1. データ統合の発展と現状

Ideal Auto のデータ統合の開発は、次の 4 つの段階を経てきました。

フェーズ 1: 2020 年 7 月に、DataX に基づいてオフラインデータ交換機能が構築されました。

フェーズ2：2021年7月にFlinkをベースとしたリアルタイム処理プラットフォームが構築されました。これら 2 つのフェーズでは、実際のデータ統合製品は存在しませんでした。

フェーズ3：2022年9月にデータ統合プラットフォームの構築を開始し、最初のデータ統合リンクを構築し、KafkaからHiveへのデータリンクを実現しました。

フェーズ4：2023年4月に、従来のリアルタイム処理機能をベースにオフライン統合機能を拡張し、バッチデータとストリームデータの統合を実現しました。

初期の頃、Ideal には統一されたデータ統合プラットフォームがなく、データ製品は多岐にわたりました。

TiDB、MySQL、StarRocks、MongoDB などから下流へのデータ転送は DataX を介して実現されます。 Kafka、Oracle などからのストリーミング特性を持つデータ転送は Flink を通じて実現されます。同時に、Hive などの一部のデータ転送は Spark SQL を記述することで実現されます。 TiDB や Oracle などの一部のデータベースでは、データベース独自のエンジンを介してデータを送信します。

上記のデータフォームに基づくと、ビジネス側では製品の使用時に次のような問題点があります。

製品機能の不足。複数のプラットフォームを切り替える必要があり、直接実装できる製品がなく、開発チームがコードを記述する必要があります。
複数の開発言語。 Flink、Spark、DataXなど、各エンジンには独自の構成があり、複数の開発言語セットと異なる開発の詳細を同時に理解する必要があります。
リソースの共有は困難です。バッチ処理とストリーム処理は異なるエンジンを使用し、異なるチームによって開発されるため、基盤となるコンピューティングリソースとストレージリソースを共有することは困難です。
リソースの使用率が低い。リソースの共有が難しいからこそ、リソースの利用率の低下と不均衡という別の問題が発生します。たとえば、リアルタイムコンピューティングクラスターは長時間実行されるタスクであり、コンピューティングリソースが不足することがよくありますが、ストレージリソースは基本的に使用されません。

ビジネスの問題点に基づいて、3 つの主要なニーズをまとめました。

1 つ目は、さまざまな異種データソース間の異なる伝送エンジン間の違いを隠す統合プラットフォームです。

2 つ目は、コンピューティングエンジンを統合し、1 セットのエンジンでバッチとストリーミングを実装することです。

3 つ目は、ストレージとコンピューティングの分離です。これにより、ビジネスニーズに応じて、コンピューティング層とストレージ層で独立した柔軟なスケーリングが可能になります。

上記の要件を満たすために、コンピューティングエンジンとして Fink を使用することを選択しました。バッチストリーム統合コンピューティングエンジンにより、バッチ処理とストリーミング処理をシームレスに切り替えることができます。同時に、K8s に基づく Flink のクラウドネイティブ機能は、コンピューティングリソースとストレージリソースの柔軟なスケーリングの実現にも役立ちます。

当社の製品は多くのビジネスアプリケーションで使用されています。 Oracle、MySQL、TiDBなどのサーバー側ログデータに接続し、binlogで業務データを下流に送信します。同時に、車両、クラウド、工場内のいくつかの埋設ポイントや信号に関するデータも統合プラットフォームを通じて送信されます。

コンピューティングの面では、伝送機能の面では、ストリーミングとバッチデータ処理の変換を実現し、並列読み取りと並列書き込み機能をサポートし、異種データソースからの異なるタイプのデータを変換する機能を備えています。ビジネスユーザーは、さまざまな製品の詳細を理解する必要がなくなりました。製品の運用・保守機能としては、タスク管理、権限制御、監視・アラーム、ログ収集など、タスクのライフサイクル全体をカバーします。最終的には、さまざまな下流のデータストレージに落ちます。

2. データ統合の実装

1. データ統合プラットフォームのアーキテクチャ

ストレージ層: JuiceFS+BOS が使用され、K8s ノードの一部のローカルストレージ機能を使用して、コンピューティングエンジンに対応するストレージ機能が提供されます。

コンピューティング層: Flink カーネルに基づいて、さまざまなコネクタが拡張され、最終的に標準化されたイメージにカプセル化されます。イメージは Flink Operator を通じて呼び出され、タスクを K8s クラスターに送信します。同時に、Flink の履歴サービスも搭載されており、例外発生時やタスク終了時のタスクの履歴ステータスを分析できます。 Flink Operator は Flink 上で定義された CRD であり、Kubernetes API サーバーを通じて外部に標準化された API を提供します。中間層は上部にカプセル化されており、K8s API サーバーを使用して製品の標準化された API をカプセル化し、各層の依存関係を保護します。また、タスクのオーケストレーションとタスクのライフサイクル管理も行います。

2. 設計モデル

データ統合の設計モデルを下図に示します。データ転送の変換は、さまざまなソースプラグインとシンクプラグインを定義することによって実現されます。

以下にカプセル化された API の場合、ユーザーは、変換プロセスを完了するために、送信するソースとデータコンテンツ、および書き込むシンクを定義するだけで済みます。

たとえば、TiDB、OceanBase、Hive、Kafka などのシンクへのリンクがすでにあるとします。新しい MySQL コネクタが追加されると、このプラグインセットのデータ転送機能のセットが作成されます。このようにして、さまざまなシナリオにデータを迅速に実装できます。

3. 典型的なシナリオ

オフライン統合シナリオでは、まずライブラリテーブル間の関連関係を取得します。増分および完全なデータ同期は、統合されたスケジューリングプラットフォームを通じて実行されます。

これまで、OceanBase から Hive へのリンクを使用する場合、OB 自体にタイムアウト設定があるため、データ量が多いと OB がタイムアウトになることがよくありました。私たちの解決策は、まず OB のデータ構造を取得し、主キーとパーティション選択シャーディングフィールドを分析し、このフィールドの最大値と最小値、およびこのバッチ内のデータ量を計算し、これら 3 つの情報を使用して、このデータを取得するサイズを合理的に設定することです。その後、Flink はサイズに基づいて並列にプルできます。毎回取得するデータの量が大きすぎないことを確認し、データタイムアウトの問題を解決します。

これまで、リアルタイム伝送リンクを開発するには、ユーザーは複数のプラットフォームにまたがって作業する必要があり、長い開発プロセスが必要でした。さらに、ユーザーはテーブルを手動で作成する必要があり、開発が複雑になります。

データ統合プラットフォームを使用すると、上記の一連の手動プロセスが不要になります。統合プラットフォーム上でソースとシンクを構成することで、データフローを実現できます。

Hive テーブルの場合、多くの場合、データパーティションが存在します。 Hive パーティションを生成する方法はいくつかあり、データ、処理、またはメタデータの時間に基づいてパーティション分割できます。メタデータ時間パーティショニングの利点は、遅延なくデータを消費する場合、パーティションのデータは基本的に同じ Hive パーティションに書き込まれるため、小さな Hive ファイルが大量に生成されるのを回避できることです。同時に、Kafka の自動パーティション認識機能が有効になります。たとえば、Kafka データが劇的に増加すると、Kafka トピックのパーティションが増加し、自動認識が非常に必要になります。

上の図は、Oracle 入力ストリームのシナリオを示しています。 Flink CDC 機能の助けを借りて、フルデータステージでは、フルデータを読み取るために複数の並列度が設定されます。データ全体が読み取られると、Flink は自動切り替え機能によって増分モードに切り替わります。増分モードでは、増分データを読み込むタスクマネージャーの 1 つが選択されます。

4. 異種データソース

データベースの種類によってサポートされるデータ型が異なり、このプロセス中にどの型をどの型に変換するかを覚えておくのは困難です。したがって、データ統合プラットフォームでは、データソースタイプを Flink タイプにマップし、データターゲットタイプも Flink タイプにマップします。最終的には、すべてが Flink 型を通じて均一に処理および変換されます。ユーザーはマッピングプロセスに注意を払う必要はありません。

5. SQLフィルタ条件

この変換プロセス中に、いくつかの一般的な where 条件をフィルタリングする必要があります。よく使われる関数をいくつか紹介します。

3. データ統合クラウドネイティブの実装

K8s クラウドネイティブソリューションの実装では、主に 4 つの重要なポイントが考慮されます。以下で、それらを 1 つずつ紹介します。

1. ソリューションの選択

選択に関しては、タスク管理に Flink Operator を使用することを選択します。まず、Flink Operator はクラスターを簡単に管理できます。 K8s アプリケーションをカプセル化し、API を拡張してアプリケーションインスタンスを構成および作成できます。宣言的にコミットします。また、統合イングレスも搭載されており、Flink の Web UI を設定したり、操作中に Web UI を通じてタスクのステータスを監視したり、操作ログを表示したりすることができます。 Flink Operator は、アプリケーションの実行と一時停止、ステートフルおよびステートレスアプリケーションのアップグレード、チェックポイントの時間指定トリガーと管理など、ジョブのライフサイクル全体の管理を実装します。ロールバックも可能です。

上図はFlink Operatorの処理プロセスを示しています。まず、プラットフォームに Flink Operator を登録します。つまり、K8s クラスターに Flink デプロイメント用の CRD を作成します。この CRD を使用して、対応するリソースを作成できます。 yaml ファイルが K8s クラスターに送信されると、K8s API は CRD を呼び出して FlinkDeployment を作成します。次に、Flink Operator は Flink Deployment と対応する TaskManager を作成します。同時に、Operator は FlinkDeployment のステータスを監視します。FlinkDeployment は実際には JobManagerPod のステータスを監視し、Operator 内で更新します。タスクが失敗した場合は、タスクの再開が試行されます。

2. 状態判定とログ収集

さまざまなタスクのステータスが FlinkDeployment に記録されています。 Flink K8s API サーバーをウォッチ方式で監視することで、各タスクイベントのステータスをキャプチャできます。同時に、各 JobManager および TaskManager ポッドも監視し、ポッドのステータスと名前をログのタイトルとして使用します。タスクの実行ステータスがすでにあるのに、なぜポッドのステータスを収集する必要があるのですか?リアルタイムタスクは永続的なタスクであるため、何らかの理由により、特定の時点でポッドが終了する可能性があります。デッドポッドの場合、すべてのログのステータスを確認する必要はありません。ポッドのステータスをマークすることで、ログが有効か無効かを説明することができます。各 K8s ノードにエージェントがデプロイされ、エージェントは各ポッドのログをダウンストリーム変換ログとして収集します。

上図は状態遷移関係を表しています。失敗、完了、キャンセル、一時停止の 4 つの状態は、最終的なタスク完了の状態タイプです。障害が発生した場合、ダウンストリームは対応するアラームを発行します。

3. 監視と警報

各タスクのアラームを設定する方法をユーザーに提供します。ユーザーがタスクを開始すると、タスクは対応するインジケーターを Prometheus に報告します。 Prometheus は定期的にそれらを収集して計算します。アラームインジケーターが作動しない場合は、サイレント状態になります。アラームインジケーターがトリガーされると、対応するユーザーにアラームが送信されます。ユーザーはアラームに基づいて適切なアクションを実行できます。

4. 共有ストレージ

JuiceFS は共有ストレージに使用されます。各ポッドは、ローカル CSI をマウントしてローカルストレージディレクトリを形成し、JuiceFS をローカルにマウントします。

Flink タスクは定期的なチェックポイントを実行する必要があり、これは JuiceFS に保持されます。タスクの実行中、Flink は再起動戦略を構成します。オペレーターにはいくつかの再起動戦略もあります。タスクが異常な場合、タスクは再起動されます。再起動すると、再起動のための最新のチェックポイントが見つかります。さらに、Flink Operator はタスクのステートレスおよびステートフルアップグレードを実装できます。アップグレード中に YAML ステータスが変更されると、最新のチェックポイントが見つかり、タスクが再開されます。

Flink 操作中のステータス情報とアーカイブ情報も JuiceFS に記録され、Flink の履歴を通じて表示できるようになります。