Delta Lake と MinIO を組み合わせてマルチクラウド データ レイクを実現

Delta Lake は、Lakehouse アーキテクチャのオブジェクト ストレージ上にデータ レイクを構築するために使用されるオープンソース ストレージ フレームワークです。 Delta Lake は、ACID トランザクション、スケーラブルなメタデータ処理、統合ストリーミングおよびバッチ データ処理をサポートしています。 Delta Lake は、Apache Spark アプリケーションに信頼性、一貫性、およびスケーラビリティを提供するために一般的に使用されます。 Delta Lake は、MinIO などの既存のデータ レイク ストレージの上で実行され、Apache Spark API と互換性があります。

Delta Lake のオリジナルの論文 ( Delta Lake: High-Performance ACID Table Storage over Cloud Object Stores ) では、Delta Lake がクラウド オブジェクト ストレージ用にどのように構築されたかについて説明しています。 Vertica が外部テーブルに対する Delta Lake の使用をテストしたとき、MinIO に依存していました。 HPE Ezmeral Runtime Enterprise の顧客はMinIO で Delta Lake を実行します。 MinIO は、分散モードとスタンドアロン モードの両方で、すべての I/O 操作に対して厳密な書き込み後読み取りおよび書き込み後リストの一貫性モデルに従っており、Delta Lake ワークロードを実行することが広く認められているため、Delta Lake の耐久性要件をサポートしています。

多くの組織は、大規模な構造化データセット、半構造化データセット、非構造化データセットを格納するために、MinIO や AWS S3 などのクラウド ネイティブ オブジェクト ストアに依存しています。各テーブルは、Parquet または ORC であるオブジェクトのセットとして格納され、パーティションに配置されます。大きなファイルに対するクエリは、基本的には高速に実行されるスキャンです。

Delta Lake がないと、より複雑な Spark ワークロード、特にデータの変更、追加、削除を行うワークロードは、マルチユーザー/マルチアプリの負荷が重い場合、パフォーマンスと正確性が課題に直面します。複数オブジェクトの更新はアトミックではなく、クエリは分離されていません。つまり、1 つのクエリで削除が実行されると、元のクエリが各オブジェクトを更新するため、他の同時クエリでは部分的な結果が得られます。書き込みのロールバックは難しく、更新中にクラッシュするとテーブルが破損する可能性があります。本当のパフォーマンスのキラーはメタデータです。数十億または数兆のレコードを保持する Parquet ファイルである数百万のオブジェクトを含む大規模なテーブルの場合、メタデータ操作により、データ レイク上に構築されたアプリケーションが停止する可能性があります。

Delta Lake は、データベースのトランザクションの信頼性とデータ レイクの水平スケーラビリティを組み合わせるように設計されました。 Delta Lake は、MinIO などのクラウド ネイティブ オブジェクト ストア上の ACID テーブル ストレージ レイヤーを使用して、OLAP スタイルのワークロードをサポートするように構築されました。論文「Delta Lake: クラウド オブジェクト ストア上の高性能 ACID テーブル ストレージ」で説明されているように、「Delta Lake の中心的なアイデアはシンプルです。どのオブジェクトが Delta テーブルの一部であるかに関する情報を、書き込みそれ自体はクラウド オブジェクト ストアに保存される先行ログです。」オブジェクトは Parquet でエンコードされ、Parquet を理解するエンジンで読み取ることができます。複数のオブジェクトを「シリアル化された方法で同時に」更新でき、同時に高い並列読み取りおよび書き込みパフォーマンスを実現できます。ログには各ファイルの最小/最大統計などのメタデータが含まれており、オブジェクト ストア内のファイルを直接検索するよりも「桁違いに高速なメタデータ検索が可能」になります。

デルタ湖では次のものが提供されます。

• ACID 保証: Delta Lake は、データに対するすべての変更がストレージに書き込まれ、永続性のためにコミットされると同時に、ユーザーとアプリがアトミックに利用できるようにします。データ レイクには、部分的なファイルや破損したファイルはもう存在しません。
• スケーラブルなデータとメタデータの処理: Spark (または別の分散処理エンジン) を使用したすべての読み取りと書き込みは、ペタバイトまで拡張できます。他のほとんどのストレージ形式やクエリ エンジンとは異なり、Delta Lake は Spark を活用してすべてのメタデータ処理をスケールアウトし、ペタバイト規模のテーブルの数十億ファイルのメタデータを効率的に処理できます。
• 監査履歴とタイムトラベル: Delta Lake トランザクション ログには、変更の完全な監査証跡を含む、データに加えられたすべての変更に関する詳細が記録されます。データ スナップショットを使用すると、開発者は監査、ロールバック、またはその他の理由でデータの以前のバージョンにアクセスして戻すことができます。
• スキーマの強制とスキーマの進化: Delta Lake は、既存のテーブル スキーマに一致しないデータの挿入を自動的に防ぎます。ただし、テーブル スキーマは、データ構造と形式の変更に対応するために明示的かつ安全に進化させることができます。
• 削除、更新、マージのサポート:ほとんどの分散処理フレームワークは、データ レイク上でのアトミックなデータ変更操作をサポートしていません。対照的に、Delta Lake は、変更データのキャプチャ、ゆっくりと変化するディメンションの操作、ストリーミング更新/挿入などの複雑なユースケースのマージ、更新、削除の操作をサポートします。
• ストリーミングとバッチの統合: Delta Lake テーブルは、バッチ モードで、ストリーミング ソースおよびシンクとして動作する機能を備えています。 Delta Lake は、ストリーミング データの取り込みやバッチ履歴バックフィルなど、さまざまなレイテンシに対応して、リアルタイムの対話型クエリを提供します。ストリーミング ジョブは、小さなオブジェクトを低遅延でテーブルに書き込み、後でそれらをトランザクション的に結合して大きなオブジェクトにし、パフォーマンスを向上させます。
• キャッシュ:オブジェクト ストレージに依存するということは、デルタ テーブル内のオブジェクトとそのログが不変であり、マルチクラウド上のどこにローカルに存在しても安全にローカルにキャッシュできることを意味します。

レイクハウス アーキテクチャ、特に Delta Lake は、オブジェクト ストレージ上に構築されたデータ レイクに重要な新機能をもたらします。 Delta Lake は、Spark、Starburst、Trino、Flink、Hive など、ますます増え続けるアプリケーションやコンピューティング エンジンのリストで動作し、Scala、Java、Rust、Ruby、Python の API も含まれています。クラウド向けに構築された Kubernetes ネイティブ MinIO は、エッジ、データセンター、パブリック/プライベート クラウドなど、あらゆる場所でパフォーマンスと復元力に優れた安全なデータ レイク アプリケーションを実現します。

デルタレイクファイル

デルタ テーブルは、ディレクトリ (ファイル システムの場合) またはバケット (MinIO およびその他のオブジェクト ストレージの場合) にまとめて保存されるファイルのコレクションです。オブジェクト ストレージから読み書きするために、Delta Lake はパスのスキームを使用してストレージ システムを動的に識別し、対応する LogStore 実装を使用して ACID 保証を提供します。 MinIO の場合は、S3A を使用します。 「ストレージ構成 — Delta Lake ドキュメント」を参照してください。 Delta Lake に使用される基盤となるストレージ システムが、MinIO と同様にアトミックな読み取り/書き込みを同時に実行できることが重要です。

デルタ テーブルの作成とは、実際にはファイルをディレクトリまたはバケットに書き込むことです。デルタ テーブルは、Spark DataFrame を書き込み (読み取り)、 delta形式とパスを指定することによって作成 (オープン) されます。 Scala では、たとえば次のようになります。

 // Create a Delta table on MinIO: spark.range(5).write.format("delta").save("s3a://<your-minio-bucket>/<path-to-delta-table>") // Read a Delta table on S3: spark.read.format("delta").load("s3a://<your-mnio-bucket>/<path-to-delta-table>").show()

Delta Lake はテーブルごとのバケットに依存しており、バケットは通常、ファイル システム パスに基づいてモデル化されています。 Delta Lake テーブルは、データ、メタデータ、トランザクション ログを含むバケットです。テーブルは Parquet 形式で保存されます。テーブルは複数のファイルに分割できます。 MinIO は、ファイル システム スタイルのパスを使用してオブジェクトを効率的にリストするための S3 LIST をサポートしています。 MinIO は、大きな Parquet ファイルのサブセットをより効率的に読み取るために、バイト範囲リクエストもサポートしています。

MinIO は、業界をリードするパフォーマンスにより、Delta Lake テーブルに最適です。 MinIO のスケーラビリティとハイパフォーマンスの組み合わせにより、要求がどれほど厳しいかに関係なく、あらゆるワークロードを手の届く範囲に収めることができます。 MinIO は驚異的なパフォーマンスを実現します。最近のベンチマークでは、わずか 32 ノードの既製 NVMe SSD で GET で 325 GiB/s (349 GB/s)、PUT で 165 GiB/s (177 GB/s) を達成しました。 MinIO は、Delta Lake 上の最も要求の厳しいワークロードに電力を供給するのに必要なパフォーマンスを十分に発揮します。

Delta Lake バケットには多くの Parquet ファイルと JSON ファイルが含まれる可能性が高く、これはデータ レイクとして使用するために MinIO に組み込まれたすべての小さなファイルの最適化と非常によく調和します。小さなオブジェクトはメタデータとともにインラインで保存されるため、Delta Lake トランザクションのような小さなファイルの読み取りと書き込みの両方に必要な IOPS が削減されます。

ほとんどの企業は、Delta Lake にマルチクラウド機能を必要とします。 MinIO には、オンプレミス、パブリック/プライベート クラウド、エッジの場所間でデータを同期するためのアクティブ/アクティブ レプリケーションが含まれています。アクティブ/アクティブ レプリケーションにより、企業は複数地域の復元力と高速なホット/ホット フェイルオーバーを設計できます。各バケットまたは Delta Lake テーブルは、セキュリティと可用性を最大限に高めるために個別にレプリケーションを構成できます。

Delta Lake との ACID トランザクション

ACID (原子性、整合性、分離性、耐久性) トランザクションをデータ レイクに追加することは、非常に重要なことです。なぜなら、組織はデータ レイクに保存されている大量のデータをより詳細に制御できるようになり、そのため信頼性が高まるからです。以前は、Spark に依存してデータ レイクを操作していた企業にはアトミック API と ACID トランザクションが不足していましたが、現在では Delta Lake によってそれが可能になっています。データはキャプチャおよび書き込み後に更新でき、ACID のサポートにより、操作中にアプリケーションに障害が発生してもデータは失われません。 Delta Lake は、データの読み取りと書き込みのために Spark と MinIO の間の仲介者として機能することでこれを実現します。

Delta Lake の中心となるのは、ユーザーとアプリケーションによって実行されたトランザクションの順序付けされた記録であるDeltaLogです。ユーザーが Delta Lake テーブルに対して実行するすべての操作 (UPDATE や INSERT など) は、複数のアクションまたはジョブで構成されるアトミック コミットです。すべてのアクションが正常に完了すると、コミットがエントリとして DeltaLog に記録されます。いずれかのジョブが失敗した場合、コミットは DeltaLog に記録されません。アトミック性がないと、ハードウェアまたはソフトウェアに障害が発生し、データが部分的にしか書き込まれずにデータが破損する可能性があります。

Delta Lake は、操作を次の 1 つ以上のアクションに分割します。

• ファイルを追加 - ファイルを追加します

• ファイルを削除 - ファイルを削除します

• メタデータの更新 - テーブルの名前、スキーマ、またはパーティショニングへの変更を記録します。

• トランザクションの設定 - ストリーミング ジョブがデータをコミットしたことを記録します

• コミット情報 - 操作、ユーザー、時間などのコミットに関する情報

• プロトコルの変更 - DeltaLog を最新のソフトウェア プロトコルに更新します

  
  

見た目ほど複雑ではありません。たとえば、ユーザーがテーブルに新しい列を追加し、そこにデータを追加すると、Delta Lake はそれをコンポーネントのアクションに分割し、メタデータを更新して列を追加し、新しいファイルが追加されるたびにファイルを追加します。完了すると DeltaLog が記録されます。

Delta Lake は、オプティミスティック同時実行制御を利用して、特定のテーブルの複数のリーダーとライターがテーブル上で同時に作業できるようにします。オプティミスティック同時実行制御では、異なるユーザーによって行われたテーブルへの変更が競合することなく完了できることを前提としています。データの量が増えると、ユーザーが別のテーブルで作業する可能性も高くなります。 Delta Lake はコミットをシリアル化し、2 つ以上のコミットが同時に発生した場合には相互排他ルールに従います。そうすることで、Delta Lake は ACID に必要な分離を実現し、複数の同時書き込み後のテーブルは、それらの書き込みが連続して別々に行われた場合と同じように見えます。

前回の読み取り以降に変更された開いているテーブルに対してユーザーが新しいクエリを実行すると、Spark は DeltaLog を参照して、新しいトランザクションがテーブルにポストされたかどうかを判断し、それらの新しい変更でユーザーのテーブルを更新します。これにより、テーブルのユーザーのバージョンが Delta Lake のマスター テーブルと最新の操作に同期され、ユーザーがテーブルに対して競合する更新を行うことができなくなります。

DeltaLog、オプティミスティック同時実行制御、スキーマ強制 (スキーマを進化させる機能と組み合わせる) により、アトミック性と一貫性の両方が保証されます。

デルタログを詳しく見る

ユーザーが Delta Lake テーブルを作成すると、そのテーブルのトランザクション ログが_delta_logサブディレクトリに自動的に作成されます。ユーザーがテーブルを変更すると、各コミットは JSON ファイルとして_delta_logサブディレクトリに昇順で書き込まれます (つまり、 000000.json 、 000001.json 、 000002.jsonなど)。

データ ファイル1.parquetおよび2.parquetからテーブルに新しいレコードを追加するとします。そのトランザクションは DeltaLog に追加され、ファイル000000.jsonとして保存されます。後で、これらのファイルを削除し、代わりに新しいファイル3.parquetを追加します。これらのアクションは、新しいファイル000001.jsonとして記録されます。

1.parquetと2.parquetを追加した後、それらを削除しました。トランザクション ログには、両方の操作が互いに否定されている場合でも、両方の操作が含まれます。 Delta Lake はすべてのアトミック コミットを保持し、完全な監査履歴と、特定の時点でテーブルがどのように見えたかをユーザーに示すタイム トラベル機能を有効にします。さらに、 VACUUMジョブが実行されるまで、ファイルはストレージからすぐには削除されません。 MinIO のバージョン管理により、誤った削除に対するもう 1 つの保証層が提供されます。

Delta Lake と MinIO による耐久性

Delta Lake は、テーブルとトランザクション ログを永続メディアに保存することで耐久性を実現します。ファイルは上書きされないため、積極的に削除する必要があります。ストレージに書き込まれたすべてのデータ変更は、発生時にアトミックにユーザーに利用可能です。部分的なファイルや破損したファイルは過去のものになります。 Delta Lake はテーブルとログを RAM に長時間保持せず、それらを MinIO に直接書き込みます。コミット データが DeltaLog に記録され、JSON ファイルがバケットに書き込まれている限り、システムまたはジョブのクラッシュが発生した場合でもデータは持続します。

MinIO は、テーブルとそのコンポーネントが複数のメカニズムを通じて書き込まれた後の耐久性を保証します。

• イレイジャー コーディングでは、データ ファイルをデータ ブロックとパリティ ブロックに分割してエンコードし、エンコードされたデータの一部が利用できない場合でもプライマリ データを回復できるようにします。水平方向に拡張可能な分散ストレージ システムは、イレイジャー コーディングを利用して、エンコードされたデータを複数のドライブとノードにまたがって保存することでデータ保護を実現します。ドライブまたはノードに障害が発生したり、データが破損したりした場合、他のドライブやノードに保存されているブロックから元のデータを再構築できます。
• Bit Rot Protection は、破損したオブジェクトを即座にキャプチャして修復し、データ耐久性に対するこの静かな脅威を除去します。
• バケットとオブジェクトの不変性は、オブジェクトのロック、保持、その他のガバナンス メカニズムを組み合わせて、MinIO に保存されたデータを削除や変更から保護します。 MinIO に書き込まれたオブジェクトは上書きされません。
• バケットとオブジェクトのバージョニングにより、オブジェクトがさらに保護されます。 MinIO は、オブジェクトが削除された場合でも、すべてのオブジェクトのすべてのバージョンの記録を保持するため、(デルタ レイクのタイムトラベルとよく似た) 過去に戻ることができます。バージョン管理はデータ ライフサイクル管理の重要なコンポーネントであり、これにより管理者は層間でバケットを移動できます。たとえば、パフォーマンス重視のワークロードに NVMe を使用したり、バージョンに有効期限を設定してストレージ効率を向上させるためにシステムから削除したりできます。

MinIO は、暗号化を使用して Delta Lake テーブルを保護し、IAM とポリシーベースのアクセス制御の組み合わせを使用してテーブルへのアクセスを規制します。 MinIO は、転送中のデータを TLS で暗号化し、ドライブ上のデータを、AES-256-GCM、ChaCha20-Poly1305、AES-CBC などの最新の業界標準暗号化アルゴリズムを使用した詳細なオブジェクト レベルの暗号化で暗号化します。 MinIO は、ActiveDirectory/LDAP、Okta、IAM 用の Keycloak などの外部ID プロバイダーと統合します。ユーザーとグループは、Delta Lake テーブルにアクセスしようとすると、AWS IAM 互換の PBAC の対象になります。

Delta Lake と MinIO のチュートリアル

このセクションでは、単一クラスター モードを使用して MinIO でデルタ テーブルの読み取りと書き込みを迅速に開始する方法について説明します。

前提条件

1. Apache Sparkをダウンロードしてインストールします。
2. MinIOをダウンロードしてインストールします。 IP アドレス、TCP ポート、アクセス キー、および秘密キーを記録します。
3. MinIO クライアントをダウンロードしてインストールします。
4. 以下のjarファイルが必要です。 jar は、Spark マシン上の任意の場所/home/sparkなど) にコピーできます。
5. Hadoop - hadoop-aws-2.6.5.jar - Delta Lake には、S3 用の Hadoop のFileSystem API を実装するhadoop-awsパッケージのorg.apache.hadoop.fs.s3a.S3AFileSystemクラスが必要です。このパッケージのバージョンが、Spark が構築された Hadoop のバージョンと一致していることを確認してください。
6. AWS - aws-java-sdk-1.7.4.jar

Delta Lake で Apache Spark をセットアップする

Delta Lake で Spark シェル (Scala または Python) を起動し、コード スニペットを対話的に実行します。

スカラの場合:

 bin/spark-shell --packages io.delta:delta-core_2.12:1.2.1 --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

Apache Spark で Delta Lake と AWS S3 を構成する

次のコマンドを実行して、Delta Lake と S3 の MinIO サポートを備えた Spark シェルを起動します。

 bin/spark-shell \ --packages io.delta:delta-core_2.12:1.2.1,org.apache.hadoop:hadoop-aws:3.3.1 \ --conf spark.hadoop.fs.s3a.access.key=<your-MinIO-access-key> \ --conf spark.hadoop.fs.s3a.secret.key=<your-MinIO-secret-key> --conf "spark.hadoop.fs.s3a.endpoint=<your-MinIO-IP:port> \ --conf "spark.databricks.delta.retentionDurationCheck.enabled=false" \ --conf "spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension" \ --conf "spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog"

MinIO でバケットを作成する

MinIO クライアントを使用して、Delta Lake のバケットを作成します。

 mc alias set minio http://<your-MinIO-IP:port> <your-MinIO-access-key> <your-MinIO-secret-key> mc mb minio\delta-lake

MinIO でテスト Delta Lake テーブルを作成する

試してみて、Scala を使用して単純な Delta Lake テーブルを作成します。

 // Create a Delta table on MinIO: spark.range(500).write.format("delta").save("s3a://delta-lake/demo1")

Spark がテーブルを正常に書き込んだことを示す出力が表示されます。

ブラウザを開いて、アクセス キーと秘密キーを使用してhttp://<your-MinIO-IP:9001>にある MinIO にログインします。バケット内に Delta Lake テーブルが表示されます。

データ レイク上での高性能 ACID トランザクションのための MinIO と Delta Lake

MinIO と Delta Lake を組み合わせることで、企業は統合された単一の信頼できる情報源として機能するマルチクラウド データ レイクを構築できるようになります。 Delta Lake テーブルのクエリと更新機能により、企業は自社のビジネスと顧客に関する豊富な洞察を得ることができます。さまざまなグループが、自分たちの作業が安全でデータがタイムリーであることを認識しながら、独自の分析や機械学習の取り組みのために Delta Lake テーブルにアクセスします。

さらに詳しく知りたい場合は、 MinIO をダウンロードして自分の目で確認するか、パブリック クラウド上でマーケットプレイス インスタンスを起動してください。質問ありませんか？ Slackまたは [email protected] 経由で質問してください。