Iceberg テーブルを管理および最適化して、データの保存とクエリを効率的に行う方法を教えてください。

解決策

ユースケースに合った Iceberg テーブルのタイプを選択する

選択する解決策は、設定する Iceberg テーブルのタイプによって異なります。

読み取りにコピーオンライト (CoW) テーブルを使用する

このタイプの Iceberg テーブルでは、レコードの更新または削除操作を行うと、対応するデータファイルがバックエンドで書き換えられます。書き換えを行うと、とりわけ更新や削除が複数発生する場合に、パフォーマンスが低下します。書き込みではなく読み取りが多いユースケースの場合は、CoW テーブルを使用してください。

既存の Iceberg テーブルを CoW テーブルに変換するには、次の Apache Spark SQL コマンドを実行します。

spark.sql("ALTER TABLE <table-name>
SET TBLPROPERTIES ('write.delete.mode'='copy-on-write','write.update.mode'='copy-on-write')")

新しい CoW テーブルを作成するには、テーブルプロパティ 'write.delete.mode'='copy-on-write','write.update.mode'='copy-on-write' を使用します。

dataFrame.createOrReplaceTempView("tmp_<your_table_name>")

query = f"""
CREATE TABLE glue_catalog.<your_database_name>.<your_table_name>
USING iceberg
TBLPROPERTIES
('format-version'='2','write.delete.mode'='copy-on-write','write.update.mode'='copy-on-write')
AS SELECT * FROM tmp_<your_table_name>
"""
spark.sql(query)

書き込みにマージオンリード (MoR) テーブルを使用する

MoR テーブルのレコードを更新または削除すると、アクションによって新しいデータファイルが追加されます。新しく追加された削除データファイルは、読み取り中にマージされます。ただし、書き込み操作では、既存のファイルに新しいファイルを追加するだけで済みます。テーブルから読み取るよりもテーブルへの書き込みの頻度が高い場合は、MoR テーブルを選択してください。

MoR Iceberg テーブルを使用するには、次の Spark SQL コマンドを実行します。

spark.sql("ALTER TABLE <table-name>
    SET TBLPROPERTIES ('write.delete.mode'='merge-on-read','write.update.mode'='merge-on-read')")

**注:**プロパティをより適切に制御するには、Spark エンジンから Iceberg テーブルを作成するのがベストプラクティスです。AWS Glue、EMR Spark、または Amazon Athena を使用してテーブルを作成することもできます。ただし、Athena ではテーブルプロパティのサポートが限られており、MoR タイプのテーブルのみを使用します。

Iceberg テーブルを最適化する

Iceberg テーブルでは、メタデータファイル数の増加やファイルの削除などが原因で、クエリのパフォーマンスが低下することがあります。クエリの効率とデータストレージを最適化する方法をいくつか紹介します。

スナップショットを期限切れにする

Iceberg テーブルはスナップショットを保持するので、テーブルの過去の状態からデータを取得できます。これらのスナップショットは、テーブルに対して実行される書き込み操作ごとに書き込まれ、関連するスナップショット ID が新しいメタデータファイルに追加されます。時間が経つにつれて、スナップショットの数が増えると、メタデータファイルのサイズが大きくなります。このようなスナップショットの増加により、クエリのパフォーマンスが低下します。

スナップショットを期限切れにするには、次のような方法があります。

• 大きなテーブルでは、Sparkの expireSnapshots 操作を使用して、指定されたタイムスタンプより古いスナップショットを並行して期限切れにします。

      SparkActions.get()
      .expireSnapshots(table)
      .expireOlderThan(tsToExpire)
      .execute()

• または、expire_snapshots というプロシージャを使用します。詳細については、Iceberg のウェブサイトで「expire_snapshots」を参照してください。

   spark.sql("CALL glue_catalog.system.expire_snapshots('databasename.tablename',<timestamp value>)")

  前述のコードを AWS Glue ジョブ内で定期的に実行します。スナップショットの期限切れを自動化すると、データファイルの数の制限、メタデータファイルの小さい状態での維持、効率的なクエリパフォーマンスの維持が可能です。

古いメタデータファイルを削除する

**write.metadata.delete-after-commit.enabled ** テーブルプロパティを True に設定すると、テーブルがコミットされるたびに古いメタデータファイルが自動的に削除されます。write.metadata.previous-versions-max を設定して、保持する古いメタデータファイルの数を管理することもできます。

マニフェストファイルを書き換える

Iceberg テーブルは、マニフェストとマニフェストファイルを使用してすべてのデータファイルを追跡します。時間が経つにつれて、各スナップショットは多数のマニフェストファイルを参照するようになります。これらの操作はクエリの速度を低下させます。詳細については、Iceberg Web サイトでマニフェストリストとマニフェストファイルを参照してください。

マニフェストの書き換えプロシージャを使用すると、マニフェストファイルを効率的に管理できます。詳細については、Iceberg のウェブサイトで「rewrite_manifests」を参照してください。

次の Spark SQL クエリを実行します。

spark.sql("CALL glue_catalog.system.rewrite_manifests('databasename.tablename')")

データファイルを書き換える

Iceberg は、テーブルのすべてのデータファイルをメタデータファイルで管理し、追跡します。時間が経つにつれて、大量のデータファイルが蓄積されると、メタデータファイルのサイズが大きくなります。メタデータファイル内に不要なファイルや開いているファイルがあると、読み取り効率が低下します。Spark の rewrite_data_files プロシージャで、データを並行して圧縮し、読み取り効率を高めることができます。詳細については、Iceberg のウェブサイトで「rewrite_data_files」を参照してください。

次の Spark SQL コマンドを実行します。

spark.sql("CALL glue_catalog.system.rewrite_data_files(table=>'databasename.tablename')")

BINPACK または SORT ストラテジーを使用して、ユースケースに合わせてデータファイルを書き直してください。詳細については、Iceberg のウェブサイトで「BINPACK」と「SORT」を参照してください。

BINPACK: これは最も安価であり、最速のアプローチでもあります。小さなファイルを大きなファイルに結合し、出力ファイルの総数を減らします。レコードの順序が乱れたり、データがシャッフルされたりすることはありません。これがデフォルトのオプションです。

CALL catalog.system.rewrite_data_files(
  table => 'test_table',
  strategy => 'binpack',
  options => map(
    'rewrite-job-order','bytes-asc',
    'target-file-size-bytes','<set-afile-size>',
    'max-file-group-size-bytes','<max-group-size>' -- 10GB
  )
)

SORT: SORT ストラテジーでは、ファイルを圧縮しながらデータをソートします。この方法は、隣接するレコードを比較する集計関数の多く (たとえば、min 関数や max 関数) を実行する場合に便利です。

CALL catalog\_name.system.rewrite\_data\_files(
    table => 'databasename.tablename',
    strategy => 'sort',
    sort\_order => 'id', --can be any column
    options => map('rewrite-all','true')
)

孤立ファイルを削除する

孤立ファイルは、どのメタデータファイルでも参照されません。詳細については、Iceberg のウェブサイトで「remove_orphan_files」を参照してください。

孤立ファイルを削除するには、以下のように remove_orphan_files コマンドを実行します。

spark.sql("CALL glue_catalog.system.remove_orphan_files(table=>'databasename.tablename')")

注: スケジュールされたジョブを実行してメンテナンスアクティビティを管理するのがベストプラクティスです。単一の AWS Glue ジョブを使用して、上記の Spark SQL クエリをすべて実行します。

関連情報

AWS Glue で Icebergフレームワークを使用する

Apache Icebergとは