MaxCompute:デルタテーブルに基づいて、ほぼリアルタイムのフルデータと増分データのストレージと処理のための統合アーキテクチャを構築します

背景情報

時間の影響が少ないビジネスシナリオでは、一度に大量のデータを処理する必要がある場合は、MaxComputeを直接使用してビジネス要件を満たすことができます。 ただし、MaxComputeでのほぼリアルタイムのフルデータ処理と増分データ処理の要件は、データ量の増加と幅広いビジネスシナリオに伴い増加しています。 次の図は、進化する要件を示しています。

たとえば、ほぼリアルタイムのデータインポートプロセスでは、プラットフォームエンジンがトランザクションの分離や小さなファイルの自動マージなどの機能を備えている必要があります。 全データと増分データのマージプロセスには、増分データの格納、読み取り、書き込み、およびプライマリキーの使用が必要です。 統合アーキテクチャがMaxComputeに導入される前は、次の図の3つのソリューションがこれらの複雑なビジネスシナリオで使用されています。 これらのソリューションには、次の図に示すように、コスト、使いやすさ、レイテンシ、およびスループットの点でそれぞれの欠点があります。

ビッグデータのオープンソースエコシステムでは、オープンソースデータ処理エンジンなどの一般的なソリューションを使用して、上記の問題を解決します。 これらのエンジンには、Spark、Flink、Trinoが含まれ、Apache Hudi、Delta lake、Apache Iceberg、Apache Paimonなどのオープンソースデータレイク形式と深く統合されています。 エンジンは、Lambdaアーキテクチャの問題を解決するために、オープンで統一されたコンピューティングエンジンとデータストアの上に構築されています。

統合アーキテクチャにより、ほぼリアルタイムのフルデータと増分データの保存と処理

前述の問題に対処するために、MaxComputeは、ほぼリアルタイムのフルデータおよび増分データの保存と処理のための統合アーキテクチャを提供します。 このアーキテクチャは、さまざまなデータソースをサポートし、カスタム開発ツールを使用して、完全なデータと増分データを特定のストレージサービスにインポートできます。 このアーキテクチャでは、バックエンドデータ管理サービスは、データストレージ構造を自動的に最適化および調整します。 統合計算エンジンは、ほぼリアルタイムの増分データ処理およびバッチデータ処理に使用されます。 トランザクションおよびファイルのメタデータを管理するために、統合メタデータサービスも提供される。 次の図は、統合アーキテクチャを示しています。

現在の統合アーキテクチャは、プライマリキーテーブル、リアルタイムアップサート、タイムトラベルクエリ、増分クエリ、SQLデータ操作言語 (DML) 操作、テーブルデータの自動管理および最適化を含む特定のコア機能をサポートしています。 統合アーキテクチャの仕組みと関連する操作の詳細については、「デルタテーブルの概要」および「基本操作」をご参照ください。

アーキテクチャのメリット

ほぼリアルタイムの完全および増分データストレージおよび処理のための統合アーキテクチャは、Apache HudiやApache Icebergなどのオープンソースデータレイク形式の主な共通機能をサポートし、関連するビジネスプロセス間の移行を容易にします。 Alibaba Cloudによって開発された新しいアーキテクチャとして、統合アーキテクチャは、機能、パフォーマンス、安定性、および統合の面でも際立った利点を提供します。

• 統合ストレージサービス、メタデータサービス、および計算エンジンを使用して、深く効率的な統合を実現します。 統合アーキテクチャは、コスト効率の高いストレージ、効率的なデータファイル管理、高いクエリ効率、および増分データに対するタイムトラベルクエリを提供します。

• すべてのコア機能をサポートするように設計された汎用の完全なSQL構文システムを提供します。

• さまざまな複雑なビジネスシナリオでデータをインポートするための高度にカスタマイズおよび最適化されたツールを提供します。

• MaxComputeの既存のビジネスシナリオとシームレスに統合できます。 これにより、データ移行の複雑さとリスク、およびデータストレージとコンピューティングのコストを削減できます。

• 完全に自動化されたファイル管理を実現し、読み取りおよび書き込み操作の安定性とパフォーマンスを向上させます。 このアーキテクチャは、ストレージ効率を自動的に最適化し、コストを削減します。

• MaxComputeに基づく完全マネージド型サービスを提供します。 この統合されたアーキテクチャは、追加のアクセスコストなしですぐに使用できます。 アーキテクチャを有効にするには、Deltaテーブルを作成するだけです。

• 自律的で制御可能な開発スケジュールを採用します。

ビジネスシナリオ

テーブル形式とデータガバナンス

テーブル作成

ほぼリアルタイムのフルデータと増分データのストレージと処理のための統合アーキテクチャをサポートするために、MaxComputeはDeltaテーブルを導入し、統一されたテーブルデータ形式を使用します。 統合アーキテクチャは、既存のバッチ処理ワークフローと、ほぼリアルタイムの増分データの保存および処理などの新しいワークフローのすべての機能をサポートします。 次のサンプルステートメントは、テーブルを作成するための構文を示しています。

CREATE TABLE tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING) tblproperties ("transactional"="true");
CREATE TABLE par_tt2 (pk BIGINT NOT NULL PRIMARY KEY, val STRING)  PARTITIONED BY (pt STRING) tblproperties ("transactional"="true");

CREATE TABLEステートメントを実行してDeltaテーブルを作成する場合は、プライマリキーを指定し、"transactional"="true" の設定を追加するだけです。 プライマリキーは、データ行の一意性を確保するために使用され、トランザクションプロパティは、読み取り操作と書き込み操作のスナップショット分離のためのアトミック性、一貫性、分離、および耐久性 (ACID) トランザクションメカニズムを構成するために使用されます。 テーブルの作成方法の詳細については、「テーブル操作」をご参照ください。

デルタテーブルの主なパラメータ

デルタテーブルのパラメーターの詳細については、「テーブル操作」の「デルタテーブルのパラメーター」セクションをご参照ください。 デルタテーブルの主なパラメータ:

• write.bucket.num: デフォルト値は16です。 有効値: (0, 4096) 。 このパラメータは、各パーティションテーブルまたは非パーティションテーブルのバケット数を指定します。 このパラメーターには、データが書き込まれる同時ノードの数も指定します。 パーティションテーブルのこのパラメーターの値を変更できます。 パーティションがパーティションテーブルに追加された場合、このパラメーターの設定は新しいパーティションに対して自動的に有効になります。 非パーティションテーブルのこのパラメーターの値は変更できません。

  説明

  バケットの数を増やすことで、データの書き込みとクエリの並列処理を増やすことができます。 ただし、バケットの増加も悪影響を及ぼします。 各データファイルはバケットに属します。 バケットが多いほど、ファイルは小さくなります。 これにより、ストレージコストと作業負荷がさらに増加し、読み取り効率が低下します。 次の原則に従ってバケット数を設定することを推奨します。

  • Conmmonシナリオには、非パーティションテーブルとパーティションテーブルが含まれます。データ量が1 GB未満の場合は、バケット数を4〜16に設定することをお勧めします。 データ量が1 GBを超える場合は、バケットあたりのデータサイズを128 MB〜256 MBに抑えることをお勧めします。 データ量が1テラバイトを超える場合は、各バケットのデータ範囲を500 MB〜1 GBに調整することをお勧めします。

  • 大規模なパーティションの増分テーブルに関するパーティションの推奨事項:

    テーブルに多数のパーティション (500を超える) があり、各パーティションに数十メガバイトなどの少量のデータが格納されている場合、過度に小さなファイルが生成されないように、各パーティションに1つまたは2つのバケットを設定することをお勧めします。

• acid.data.retain.hours: デフォルト値は24です。 有効値: [0, 168] 。 このパラメーターには、タイムトラベルクエリ機能を使用してクエリできる履歴データの時間範囲を指定します。 単位：時間。 タイムトラベルクエリ機能を使用して履歴データをクエリする必要がない場合は、このパラメーターを0に設定することを推奨します。 この値は、タイムトラベルクエリ機能が無効になっていることを示します。 これにより、履歴データのストレージコストを大幅に削減できます。 タイムトラベルクエリ機能を使用して168時間 (7日) を超える履歴データをクエリする必要がある場合は、MaxComputeテクニカルサポートにお問い合わせください。

  ビジネスシナリオに基づいて合理的な期間を設定することを推奨します。 このパラメーターを大きな値に設定すると、大量の履歴データが保存され、ストレージコストが高くなり、クエリの速度が低下します。

  説明

  • 設定された期間が経過すると、システムは自動的に履歴データを再要求してクリアします。 すべての履歴データがクリアされた後、タイムトラベルクエリ機能を使用して履歴データをクエリすることはできません。 回収データには、操作ログとデータファイルが含まれます。

  • 特別な場合は、purgeコマンドを実行して履歴データを強制的に消去できます。

ALTER TABLE tt2 ADD columns (val2 string);

テーブルデータ形式

上の図は、パーティションテーブルのデータ構造を示しています。 データファイルはパーティションによって物理的に分離され、異なるパーティション内のデータは異なるディレクトリに格納されます。 各パーティションのデータは、バケットの数に基づいてバケットに分割されます。 各バケットのデータファイルは別々に保存されます。 データファイルは、デルタデータファイルと圧縮データファイルに分類される。

• デルタデータファイル: 各トランザクションのデータが書き込まれた後、または小さなファイルがマージされた後に生成される増分データファイル。 増分データのほぼリアルタイムの読み取りおよび書き込みの要件を満たすために、すべての行の中間履歴データがDeltaデータファイルに格納されます。

• 圧縮データファイル: デルタデータファイルが圧縮された後に生成されるデータファイル。 圧縮データファイルには、すべての行の中間履歴データは含まれません。 複数行のレコードが同じ主キーを持つ場合、保持されるのは1行だけです。 圧縮されたデータファイルは、列指向のストレージと圧縮を使用してデータクエリを高速化します。

自動データガバナンスと最適化

• 既存の問題-小さなファイルの膨張

  デルタテーブルは、数分でほぼリアルタイムの増分データのインポートをサポートします。 高トラフィックのリアルタイム書き込みシナリオでは、特に多数のバケットが存在する場合、増分データ用の小さなファイルの数が増加する可能性があります。 これは、過度のアクセス要求、高コスト、およびデータの読み書きのための低いI/O効率などの問題を引き起こす可能性があります。 UPDATEおよびDELETE操作に大量のデータが含まれる場合、多数の冗長な中間履歴レコードが生成されます。 これにより、ストレージおよびコンピューティングコストがさらに増加し、クエリ効率が低下します。

• ソリューション-自動ガバナンスと最適化

  MaxComputeストレージエンジンは、合理的かつ効率的なテーブルデータサービスをサポートします。 このサービスは、保存されたデータを自動的に管理および最適化し、ストレージとコンピューティングコストを削減し、データ処理と分析のパフォーマンスを向上させます。

  

  上の図は、Deltaテーブルのテーブルデータサービスを示しています。 サービスには、auto sort、auto merge、auto partial compact、auto cleanが含まれます。 手動でサービスを設定する必要はありません。 ストレージエンジンサービスは、各ディメンションからデータをインテリジェントに識別して自動的に収集し、自動実行用のポリシーを構成できます。

  • 自動ソート: リアルタイムで書き込まれる行指向のAvroファイルを列指向のAliORCファイルに変換します。 これにより、ストレージコストが大幅に削減され、データ読み取りパフォーマンスが向上します。

  • 自動マージ: 小さなファイルを自動的にマージして、小さなファイルの数の増加による問題を解決します。

    主なポリシーは、ファイルのサイズ、量、書き込み時系列など、複数の側面から包括的な分析を定期的に実行し、ファイルのレベルに基づいてファイルをマージすることです。 すべてのレコードの中間履歴データは、このプロセスではクリアされません。 これにより、タイムトラベルクエリの整合性とトレーサビリティが保証されます。

  • 自動部分コンパクト: ファイルを自動的にマージし、レコードの履歴データを消去します。 これにより、レコードに対する過剰なUPDATEおよびDELETE操作による追加のストレージコストが削減され、読み取り効率が向上します。

    主なポリシーは、増分データサイズ、書き込み時系列、時間移動クエリなど、複数の側面から包括的な分析を定期的に実行し、その後、コンパクトな操作を実行することです。

    説明

    この操作は、作成時刻がタイムトラベルクエリの時間範囲を超えている履歴レコードのみを圧縮します。

  • 自動消去: 無効なファイルを自動的に消去して、ストレージコストを削減します。 自動ソート、自動マージ、および自動部分コンパクト操作が実行され、新しいデータファイルが生成された後、元のデータファイルは無効になります。 システムは自動的にリアルタイムで元のファイルを削除し始めます。 これにより、ストレージスペースを節約し、ストレージコストをタイムリーに削減できます。

  高いクエリのパフォーマンスが必要なシナリオでは、フルデータに対してメジャーコンパクト操作を手動でトリガーできます。 詳細については、「COMPACTION」をご参照ください。

  SET odps.merge.task.mode=service;
  ALTER TABLE tt2 compact major;

  説明

  この操作は、各バケット内のすべての情報を統合し、すべての履歴データを完全に消去し、列指向のAliORCファイルを生成します。 ただし、この操作では追加の実行オーバーヘッドが発生し、新しいファイルのストレージコストが増加します。 必要な場合にのみ操作を実行することを推奨します。

データ書き込み

数分でほぼリアルタイムのupserts

バッチデータ処理に関して、MaxComputeでは、増分データを新しいテーブルまたはパーティションに数分または数日でインポートできます。 次に、オフラインの抽出、変換、および読み込み (ETL) プロセスを設定し、増分データと既存のテーブルデータに対して結合およびマージ操作を実行して、新しいフルデータを生成できます。 このオフラインプロセスは待ち時間が長く、リソースとストレージスペースを消費します。

ほぼリアルタイムのフルおよび増分データの記憶および処理のための統合アーキテクチャでは、リアルタイムのアップサート・プロセスは、データ書き込みからクエリまでの待ち時間を5〜10分に維持することができる。 これは、数分でのほぼリアルタイムのデータ処理の要件を満たします。 さらに、マージ操作のための複雑なETLプロセスは必要ありません。 これにより、コンピューティングとストレージのコストを削減できます。 実際のビジネスデータ処理シナリオでは、データベース、ログシステム、メッセージキューシステムなど、さまざまなデータソースが含まれます。 MaxComputeは、Deltaテーブルへのデータ書き込みを容易にするために、DataWorks data Integrationやその他のデータインポートツールと連携して、高い同時実行性、フォールトトレランス、トランザクション送信などのシナリオにカスタム設計と開発の最適化を適用できるオープンソースのFlinkコネクタプラグインを提供しています。 これにより、低レイテンシおよび高精度を達成することができる。

上の図に示すように:

• Flinkエコシステムと互換性のあるほとんどのコンピューティングエンジンとツールを使用すると、MaxCompute Flinkコネクタを効率的に使用してDeltaテーブルにリアルタイムでデータを書き込むようにFlinkデプロイを構成できます。

• Deltaテーブルのwrite.bucket.numパラメーターを調整して、書き込みの並列処理を柔軟に設定できます。 これは、書き込み動作の高速化に役立つ。

  MaxComputeは効率的に最適化されています。 Deltaテーブルのwrite.bucket.numパラメーターをFlinkシンク並列処理の整数倍に設定した場合、MaxComputeは最高の書き込みパフォーマンスを提供します。

• Flinkの組み込みチェックポイントメカニズムを使用して、フォールトトレランスを実装できます。 これは、データ処理がexactly_onceセマンティクスに従うことを保証する。

• 同時に何千ものパーティションにデータを書き込むことができます。 これは、多数のパーティションにデータを同時に書き込むための要件を満たす。

• データは数分で表示され、読み書き操作でスナップショットの分離がサポートされます。

• 異なる環境および構成は、トラフィックスループットに悪影響を及ぼし得る。 1つのバケットの処理能力 (1メガバイト/秒) に基づいて、最大トラフィックスループットを推定できます。 MaxCompute Tunnelの場合、共有トンネルリソースグループがデフォルトで使用されます。 これは、特に激しいリソース競合が発生する場合に不安定なスループットを引き起こす可能性があります。 リソース消費にも制限が課されます。

上図では：

• 左のワークフローは、MaxComputeでのバッチデータ処理のアーキテクチャを示しています。 ほとんどの場合、増分データは時間単位または日単位で新しいテーブルまたはパーティションにインポートされます。 次に、指定されたオフラインETLプロセスがトリガーされ、増分データと既存のテーブルデータに対して結合およびマージ操作が実行され、新しいフルデータが生成されます。 このオフラインプロセスは待ち時間が長く、リソースとストレージスペースを消費します。

• 右のワークフローは、ほぼリアルタイムのフルデータと増分データのストレージと処理のための統合アーキテクチャを示しています。 MaxComputeのバッチ処理アーキテクチャと比較して、統合アーキテクチャでは定期的なデータの抽出とマージは必要なく、新しいレコードは数分でデータベースから読み取られます。 データを更新するには、Deltaテーブルを使用します。 これにより、コンピューティングとストレージのコストが最小限に抑えられます。

データクエリ

タイムトラベルのクエリ

MaxCompute Deltaテーブル計算エンジンは、タイムトラベルクエリ機能が有効になっている一般的なビジネスシナリオに適しています。 タイムトラベルのクエリ機能を有効にすると、履歴バージョンのデータをクエリできます。 この機能を使用して、ビジネスデータの履歴をバックトラックしたり、データエラーが発生したときにデータ修正のために履歴データを復元したりできます。 復元操作を直接使用して、指定した履歴バージョンにデータを復元することもできます。 次のステートメントは例を示します。

// Query the historical data with a specified timestamp.
SELECT * FROM tt2 TIMESTAMP AS OF '2024-04-01 01:00:00';
// Query the historical data that are generated 5 minutes before the query time.
SELECT * FROM tt2 TIMESTAMP AS OF CURRENT_TIMESTAMP() - 300;
// Query the historical data that is written for the last second commit.
SELECT * FROM tt2 TIMESTAMP AS OF GET_LATEST_TIMESTAMP('tt2', 2);

次の図は、タイムトラベルクエリの実行方法を示しています。

タイムトラベルクエリ文を入力すると、システムはメタデータサービスから過去のデータバージョンを解析し、データの読み取り元である圧縮データファイルとデルタデータファイルを取得し、データをマージして出力します。 圧縮されたデータファイルを使用して、クエリを高速化し、読み取り効率を向上できます。

上の図では、例としてsrcというトランザクションテーブルを使用しています。

• 図の左側は、データ更新プロセスを示す。 t1〜t5は、5つのトランザクションのタイムバージョンを表す。 5つのデータ書き込みトランザクションが実行され、5つのデルタデータファイルが生成される。 t2およびt4において、圧縮操作が実行され、c1およびc2という名前の2つの圧縮データファイルが生成される。 c1ファイルでは、中間履歴レコード (2,a) が削除され、最新のレコード (2,b) が保持される。

• t1で履歴データを照会するために、システムはデルタデータファイルd1のみからデータを読み取り、出力を返す。 t2で履歴データを照会するために、システムは圧縮データファイルc1のみからデータを読み取り、3つの出力レコードを返す。 t3で履歴データを照会するために、システムは、圧縮データファイルc1およびデルタデータファイルd3内のデータを返し、出力のためにデータをマージする。 同じルールを適用して、他の時点のクエリ結果を取得できます。 圧縮データファイルを使用すると、クエリは高速化されますが、COMPACTION操作は頻繁にトリガーされます。 ビジネス要件に基づいて適切なトリガーポリシーを選択する必要があります。

• SQL構文では、定数と共通関数を直接指定できます。 TIMESTAMP AS OF exprおよびVERSION AS OF expr関数を使用して、正確なクエリを実行することもできます。

増分クエリ

デルタテーブルの増分クエリと増分コンピューティングを最適化するために、MaxComputeは新しいSQL増分クエリ構文を設計および開発します。 増分クエリ構文の詳細については、「タイムトラベルクエリと増分クエリ」の「増分クエリのパラメーターと制限」セクションをご参照ください。 次の図は、増分クエリプロセスを示しています。.

SQL文を入力した後、MaxComputeエンジンは、照会する履歴の増分データバージョンを解析し、読み取る圧縮データファイルを取得し、ファイルデータをマージして、出力結果を返します。 上の図では、例としてsrcというトランザクションテーブルを使用しています。

• 図の左側は、データ更新プロセスを示す。 t1〜t5は、5つのトランザクションのタイムバージョンを表す。 5つのデータ書き込みトランザクションが実行され、5つの圧縮データファイルが生成される。 t2およびt4において、圧縮操作が実行され、c1およびc2という名前の2つの圧縮データファイルが生成される。

• 例えば、beginがt1-1であり、endがt1である場合、システムは、t1でデルタデータファイルd1のみからデータを読み出して出力する。 終了がt2である場合、システムは2つのデルタデータファイルd1およびd2を読み取る。 beginがt1であり、endがt2-1である場合、クエリ時間範囲はt1からt2である。 増分データは時間範囲に挿入されません。 その結果、空の行が返されます。

• コンパクション操作およびマージ操作によって生成されたc1ファイルおよびc2ファイルのデータは、出力用の新しいデータとは見なされません。

最適化されたプライマリキーベースのデータスキップ

デルタテーブルのデータ分布とインデックスは、基本的に主キー列の値に基づいて作成されます。 デルタテーブルのデータをクエリする場合、特定のプライマリキー値に基づいてデータをフィルタリングできます。 これにより、読み取るデータ量を大幅に削減し、クエリを高速化し、リソース消費を大幅に削減できます。 クエリ効率は、数百から数千倍改善され得る。 最適化された主キーベースのデータスキッピングプロセスは、指定された主キー値を持つデータを効率的に見つけて読み取るために、複数レベルの正確なフィルタリングをカバーします。 たとえば、Deltaテーブルには100万件のデータレコードが含まれます。 プライマリキー値に基づいてテーブルデータをフィルタリングできます。 フィルタリングの後、10,000のデータレコードのみがテーブルから読み取られる必要があり得る。 次の図は、データ照会プロセスを示しています。