Data Lake Formation:Iceberg テーブルのメンテナンス

前提条件

バージョン要件

• Spark 4.7.0 以降。

必要なカタログ権限付与の完了:

• Serverless Spark からの DLF へのアクセス

• EMR on ECS Spark からの DLF へのアクセス

依存関係のダウンロード

• iceberg-spark-runtime-3.5_2.12-1.10.1.jar

• iceberg-aws-bundle-1.10.1.jar

• bennett-iceberg-plugin-2.8.0.jar

• guava-31.1-jre.jar

依存関係の構成

Serverless Spark

4つの依存関係 JAR パッケージを Object Storage Service (OSS) にアップロードし、次の Spark 構成をセッションに追加します。

• spark.emr.serverless.excludedModules: iceberg である必要があります。

• spark.emr.serverless.user.defined.jars: 依存関係 JAR の OSS パス。

• spark.sql.extensions: org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions に設定します。

spark.emr.serverless.excludedModules iceberg
spark.emr.serverless.user.defined.jars   oss://${bucket}/iceberg-spark-runtime-3.5_2.12-1.10.1.jar,oss://${bucket}/iceberg-aws-bundle-1.10.1.jar,oss://${bucket}/bennett-iceberg-plugin-2.8.0.jar,oss://${bucket}/guava-31.1-jre.jar
spark.sql.extensions                    org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions

Spark プロシージャの使用

Iceberg は、テーブルメンテナンスのためにいくつかの Spark プロシージャを提供します。これらのプロシージャは、CALL ステートメントを使用して Spark SQL で実行できます。

コンピュートエンジンで DLF カタログ (例: dlfCatalogName) を使用する場合は、ネイティブ Iceberg モード (dlfCatalogName_iceberg) を有効にします。

-- 1. ドライラン
-- 実際に削除せずに、削除されるファイルをリストします。
CALL dlfCatalogName_iceberg.system.remove_orphan_files(
  table => 'db.table_name',
  prefix_listing => true,     -- プレフィックスリスト機能を使用します。これにより、OSS や S3 などのオブジェクトストレージサービスでのスキャンパフォーマンスが大幅に向上します。
  dry_run => true             -- 削除をドライランし、削除されるファイルのリストを返します。
);

-- 2. デフォルトのクリーンアップ
-- 3日以上前の孤立ファイルを削除します。
CALL dlfCatalogName_iceberg.system.remove_orphan_files(
  table => 'db.table_name',
  prefix_listing => true
);

-- 3. 時間と同時実行数の設定
-- 指定されたタイムスタンプより古い孤立ファイルを削除し、並行削除を有効にします。
CALL dlfCatalogName_iceberg.system.remove_orphan_files(
  table => 'db.table_name',
  older_than => TIMESTAMP '2024-01-01 00:00:00',   -- 目的のタイムスタンプに置き換えます。
  dry_run => false,
  max_concurrent_deletes => 4,                   -- 同時削除スレッド数。
  prefix_listing => true
);

-- 1. 時間ベースのクリーンアップ
-- 指定されたタイムスタンプより古いスナップショットを削除します。
CALL dlfCatalogName_iceberg.system.expire_snapshots(
  table => 'db.table_name',
  older_than => TIMESTAMP '2024-01-01 00:00:00'     -- 目的のタイムスタンプに置き換えます。
);

-- 2. 最新のスナップショットを保持
-- 経過時間に関係なく、最新の N 個のスナップショットを保持します。
CALL dlfCatalogName_iceberg.system.expire_snapshots(
  table => 'db.table_name',
  retain_last => 5
);

-- 3. 特定のスナップショットの削除
-- ID リストで指定されたスナップショットのみを削除します。
CALL dlfCatalogName_iceberg.system.expire_snapshots(
  table => 'db.table_name',
  snapshot_ids => ARRAY(123456789, 987654321)      --目的のスナップショット ID に置き換えます。
);

-- 1. デフォルトの書き換え
-- デフォルトの binpack ストラテジーを使用してファイルを書き換えます。
CALL dlfCatalogName_iceberg.system.rewrite_data_files('db.table_name');

-- 2. Binpack ストラテジーの指定 (シンプルなビンパッキングアルゴリズムを使用して小さなファイルをより大きなファイルに圧縮します。これにより、最低限のオーバーヘッドと最速のパフォーマンスが提供されます。)
CALL dlfCatalogName_iceberg.system.rewrite_data_files(
    table => 'db.table_name',
    strategy => 'binpack'
);


-- 3. フィルター条件による書き換え
-- 特定のパーティションまたは条件に一致するデータのみを書き換えます。
CALL dlfCatalogName_iceberg.system.rewrite_data_files(
    table => 'db.table_name',
    where => 'date >= "2024-01-01"'           --'date' はテーブル内の列名の例です。必要に応じてフィルター条件を変更してください。
);

-- 4. カスタムファイルサイズオプション
-- ターゲットファイルサイズ (512 MB) と書き換えの最小入力ファイルサイズしきい値 (128 MB) を明示的に設定します。
CALL dlfCatalogName_iceberg.system.rewrite_data_files(
    table => 'db.table_name',
    options => map(
        'target-file-size-bytes', '536870912',
        'min-file-size-bytes', '134217728'
    )
);