Best practices of EMR-StarRocks and Flink in real-time writing scenarios in sink volume

EMR-StarRocks の概要

Alibaba Cloud EMR は年初に StarRocks サービスをリリースしました。StarRocks は次世代の超高速フルシーン MPP (Massively Parallel Processing) データウェアハウスであり、高速かつ統合された分析体験の構築に専念しています。EMR StarRocks には以下の特徴があります。

• MySQL プロトコルと互換性があり、MySQL クライアントや一般的な BI ツールを使用して StarRocks と連携しデータを分析できます

• 分散アーキテクチャ：

• データテーブルを水平分割し、複数レプリカで保存

• クラスターサイズを柔軟にスケーリングでき、10 PB のデータ分析をサポート

• MPP フレームワークをサポートし、並列計算を高速化

• 複数レプリカをサポートし、弾力的なフォールトトレランスを提供

• ベクトル化エンジンと CBO をサポート

• 弾力的なスケールアウトとスケールインをサポート

• ディテールモデル、集約モデル、プライマリキーモデル、更新モデルをサポート

詳細については、https://help.aliyun.com/document_detail/405463.html を参照してください。

Flink-CDC の概念紹介

CDC の正式名称は Change Data Capture (変更データキャプチャ) で、データ同期、データ分散、データ収集などのシナリオに対応します。Flink CDC は主にデータベースの変更を対象とし、上流のデータとスキーマの変更を下流のデータレイクやデータウェアハウスに同期できます。2020 年 7 月に Flink CDC プロジェクトは最初のコミットが行われ、翌年 8 月に Flink コミュニティは CDC 2.0 をリリースしました。2 年間の改良を経て、商用利用において非常に成熟しています。この記事では MySQL CDC を例に、StarRocks + Flink CDC のリアルタイムウェアハウジングにおいてユーザーが直面する課題と、Flink および StarRocks レベルでの対応する最適化とソリューションを紹介します。

CDC を使用して MySQL テーブルから StarRocks テーブルにデータをインポートするには、まず StarRocks 側に MySQL のデータを受け入れるターゲットテーブルを作成し、次に Flink 上で MySQL テーブルと StarRocks テーブルのマッピングをそれぞれソーステーブルと結果テーブルとして定義してから、INSERT INTO sink_table SELECT FROM source_table 文を実行します。INSERT INTO を実行すると CDC タスクが生成されます。CDC タスクはまずソーステーブルの全データをターゲットテーブルに同期し、その後 Binlog に基づいて増分データを継続的に同期します。1 つのタスクでデータの完全データ同期と増分同期を完了できるため、ユーザーにとって非常に便利です。ただし、利用過程においていくつかの課題も見つかりました。

リアルタイム書き込みシナリオにおけるユーザーの課題

SQL 開発の膨大な作業量

データウェアハウス構築が未完了の新規ビジネスや、StarRocks をベースにした OLAP プラットフォームの構築を始めたばかりのユーザーにとって、MySQL のデータを受け入れるための StarRocks テーブル作成が最初のステップとなります。複雑なビジネスでは、MySQL 側に数十から数百のテーブルがあり、各テーブルに数十のフィールドが存在することがよくあります。これらに対応する StarRocks テーブルの CREATE 文をすべて記述するのは膨大な作業量です。これが最初の課題であり、StarRocks のテーブル構築に多くの作業が必要です。

Flink のフィールドデータ型マッピングが複雑でエラーが発生しやすい

StarRocks でのテーブル作成は最初のステップに過ぎません。テーブル作成後、CDC タスクを開始するために、Flink 側でも MySQL に対応するソーステーブルと StarRocks に対応する結果テーブルを作成する必要があります。Flink のテーブル作成時には、各フィールドと MySQL および StarRocks の間のフィールド型マッピングを厳密に遵守する必要があります。数十から数百のテーブルがあり、各フィールドごとに Flink に対応する型マッピングを見つける必要があるため、開発者にとって非常に辛い作業です。したがって、2 番目の課題は、上流・下流のテーブルと Flink フィールド間のデータ型マッピングが複雑でエラーが発生しやすいことです。

スキーマ変更操作が煩雑

3 番目の課題はビジネスデータのスキーマ変更に起因します。Fivetran によると、企業の約 60% は毎月データスキーマの変更があり、30% は毎週変更があるとのことです。MySQL テーブルのフィールド追加、削除、修正について、ユーザーは CDC タスクに影響を与えずにスキーマ変更を下流の StarRocks に同期させたいと考えています。現在の一般的な解決策は、StarRocks と MySQL のスキーマを手動で変更し、Flink 側の結果テーブルとソーステーブルの構造を変更してから、手動でタスクを停止した後にセーブポイントを指定してタスクを再起動する必要があります。スキーマ変更の操作は煩雑であり、3 番目の課題はこれを自動化できないことです。

データ同期タスクのリソース消費が大きい

4 番目の課題は、テーブル数が多くリアルタイムの増分データ量が多いシナリオにおいて、CDC タスクが大量のメモリと CPU リソースを消費することです。コスト削減のため、ユーザーはリソース使用率を可能な限り最適化したいと考えています。

次に、EMR-StarRocks が Flink との深い統合において提供している最適化と、これらの課題に対するソリューションを見ていきましょう。

CTAS&CDAS

EMR-StarRocks と Flink チームがリリースした CTAS&CDAS 機能は、主に最初の 3 つの課題に対して開発されたソリューションです。CTAS&CDAS を通じて、1 つの SQL 文で StarRocks テーブル作成、Flink-CDC タスク作成、スキーマ変更のリアルタイム同期など、複数の複雑な操作が必要だったタスクを完了でき、開発と運用保守の作業量を大幅に削減できます。

CTAS の紹介

CTAS の正式名称は CREATE TABLE AS で、構文構造は以下の通りです。

CTAS の構文構造から、クラスター情報とデータベース情報に加え、"starrocks.create.table.properties" という特別な構成があることがわかります。これは MySQL と StarRocks でキータイプ、パーティション、バケット数などの特殊構成においてテーブル構造が異なるため、StarRocks テーブル作成文のフィールド定義以降の内容を引き受けるものです。

ユーザーがより迅速にテーブルを作成できるよう、シンプルモードも用意されています。構成方法は以下の通りです。

シンプルモードを有効にすると、デフォルトでプライマリキーモデルが使用され、MySQL のプライマリキーがデフォルトのプライマリキーとして使用され、バケット分割はデフォルトでハッシュ (プライマリキー) が使用されます。これにより、ユーザーはシンプルモードを有効にして CTAS 文を使用する際、元の MySQL テーブルのフィールド、フィールド名、プライマリキーなどを意識する必要がなく、テーブル名を知るだけで効率的に SQL を記述できます。

CTAS の原理

図に示すように、CTAS 文を実行すると、まず Flink が MySQL ソーステーブルと同じスキーマのターゲットテーブルを StarRocks に自動作成し、次に Flink 内で MySQL と StarRocks テーブルのソースとシンクのマッピングを確立してから、CDC タスクを開始します。このタスクはソーステーブルのデータをターゲットテーブルに同期し、実行時に MySQL ソーステーブルから送信されるデータのスキーマ変更を監視して、スキーマ変更を StarRocks ターゲットテーブルに自動同期します。CTAS 機能は実際に 1 つの SQL で、手動で記述・実行する必要があった多数の操作を完了します。

次に CTAS の実装原理を説明します。CTAS の実装は主に Flink CDC、Flink Catalog、Schema Evolution に依存しています。Flink の CDC 機能は前述の通りです。Catalog 機能により Flink は StarRocks 内のすべてのデータベースとすべてのテーブルスキーマを認識し、DDL 操作を実行できます。Schema Evolution 機能はデータスキーマの変更を検出・記録することで実現されます。たとえば、MySQL で列が追加された場合、CTAS タスクは MySQL の DDL 変更を検知しても即座に下流の StarRocks に列を追加するのではなく、新しいスキーマを使用する最初のデータが処理される際に新旧データスキーマの差分を比較して ALTER TABLE ADD COLUMN 文を生成し、StarRocks に列を追加します。StarRocks のスキーマ変更が完了してから初めて新しいデータが下流にプッシュされます。

CDAS の紹介

CDAS は CTAS の構文シュガーです。CDAS 文を使用することで、MySQL のデータベース全体を同期でき、1 つの Flink ジョブを生成します。ソースは MySQL のデータベース、ターゲットは StarRocks の対応する複数テーブルです。

単一の SQL で複数テーブルのスキーマと CDC タスクを生成するため、シンプルモードを統一して使用する必要があります。実際の利用過程では、データベース内の一部のテーブルは同期不要で、一部のテーブルは属性構成をカスタマイズしたい場合があるため、INCLUDING TABLE 構文を使用してデータベース内の特定テーブルのみを選択して CDAS 操作を行えます。属性構成のカスタマイズが必要なテーブルについては、CTAS 文を使用して操作します。

重要な特徴

CTAS&CDAS の重要な機能には以下が含まれます。

• 同一ジョブで複数の CDC タスクを実行でき、大量のメモリと CPU リソースを節約できます。

• ソース統合をサポートします。CDAS を使用してデータ同期を行う場合、1 つのジョブですべてのテーブルの同期タスクを管理し、すべてのテーブルのソースを自動的に 1 つに統合して MySQL 側の同時読み取り負荷を軽減します。

• サポートされるスキーマ変更タイプには、列の追加、列の削除、列名の変更が含まれます。ここで注意すべきは、現在サポートされている列の削除操作は、対応フィールドの値を null に設定することで実現されている点です。たとえば、上流の MySQL テーブルでフィールドが削除されても、Flink がデータスキーマの変更を検出した後、StarRocks の対応する列は削除されず、StarRocks へのデータ書き込み時に対応フィールドの値が null に設定されます。列名の変更操作も同様に、新しい列を追加して新しいデータでは元の列の値を空白にすることで実現されます。

Connector-V2 の紹介

Connector-V2 は 4 番目の課題を解決するために開発されました。Flink を介した StarRocks へのデータインポート時のメモリ消費を削減し、タスクの安定性を向上させます。

図に示すように、V1 バージョンでは Exactly Once を保証するために、Checkpoint 期間中のすべてのデータを Flink の Sink オペレータのメモリ内に保持する必要がありました。Checkpoint 間隔は短すぎることができず、単位時間あたりのデータフローも予測できないため、メモリリソースの深刻な消費を引き起こすだけでなく、OOM による安定性の問題も頻繁に発生していました。

V2 バージョンでは 2 フェーズコミット機能を通じてこの問題を解決しています。2 フェーズコミットとは、データのコミットを 2 つの段階に分割することです。第 1 段階はデータ書き込みタスクのコミットで、データ書き込み段階ではデータは非表示であり、複数回のバッチ書き込みが可能です。第 2 段階はコミット段階で、複数バッチで書き込まれたデータを Commit リクエストを通じて同時に表示可能にします。StarRocks 側は Begin、Prepare、Commit などのインターフェイスを提供し、複数のデータ書き込みリクエストを同一トランザクションとしてコミットすることをサポートし、同一トランザクション内のデータ一貫性を保証しています。

Transaction インターフェイスの呼び出し方式により、Flink 側で大量のデータを蓄積して一度に送信する方式から、小バッチで継続的にコミットする方式へ改善できます。Exactly Once を保証しつつ、データバッファー保存のための Flink 側のメモリ消費を大幅に削減し、Flink タスクの安定性も向上させています。

StarRocks + Flink の大量データ取り込みにおける実践

CDAS 機能を使用して、広告分析ビジネスにおける MySQL から Flink へのデータのリアルタイム同期を完了させています。

以前、このビジネスは主にクローズドソースのデータウェアハウスを OLAP 分析に使用していました。データ量の増加に伴い、単一テーブルクエリと複数テーブル結合のシナリオで大きなボトルネックが生じ、クエリ時間が分単位の許容できないレベルに達しました。そこで StarRocks をデータ分析に採用し、対応するシナリオで非常に優れたパフォーマンスを発揮しました。

集約のビジネスシナリオでは、運用メタデータに関連する数十の小規模テーブルを StarRocks 内で CDAS を使用してリアルタイム同期し、データ量が多い複数の詳細テーブルはオフラインインポート方式で日次更新しています。CDAS は主にデータ更新とスキーマ変更が頻繁な小規模テーブルとディメンションテーブルに使用しています。ビジネスクエリ時に、これらのリアルタイム更新テーブルとオフラインデータテーブルを結合します。小規模テーブルのリアルタイム更新、大規模テーブルのオフライン更新、大小テーブルの結合クエリを通じて、リアルタイム性、コスト、インポートとクエリパフォーマンスのトレードオフを実現しています。ビジネスはデータの正確性に対する要求が高いため、Flink の Checkpoint メカニズムを通じて Exactly-once セマンティクスを使用し、データの欠落や重複がないことを保証しています。