Iceberg テーブルを構築する
使用するタイミング
- レイクハウス上で Apache Iceberg テーブルを作成・保守するとき。
- パーティショニングを選択するとき、または再書き込みなしでパーティション/スキーマを進化させるとき。
- スナップショット、タイムトラベル、コンパクション、小ファイルを管理するとき。
- Delta 固有の作業には使用しないでください（
engineering-databricks-pipelinesを使用）。
ワークフロー
- [ ] クエリのフィルタ列でパーティション分割し、隠しパーティション変換を使用する
- [ ] 冪等な upsert には MERGE を使用する
- [ ] コンパクション (rewrite_data_files) をスケジュールして小ファイルを修正する
- [ ] 古いスナップショットを期限切れにし、孤立ファイルを削除してメタデータ/ストレージを制御する
- [ ] フィールド ID でパーティション/スキーマを進化させる（データ再書き込みなし）
- フィルタ列でパーティション分割 — 隠しパーティション変換（
days(ts)、bucket(N, id)）を使用するため、ユーザーが導出パーティション列を追加しなくてもクエリがプルーニングされる。 - 冪等な書き込み —
MERGE INTOをビジネスキーでキーとして使用する。 - 定期的にコンパクション — ストリーミング/小バッチ書き込みは多くの小ファイルを作成するため、
rewrite_data_filesで読み取りパフォーマンスを回復する。 - メタデータを維持 — 古いスナップショットを期限切れにし、孤立ファイルを削除する。さもないとスナップショット履歴とストレージが無制限に増大する。
- 自由に進化 — Iceberg は列とパーティションを ID で追跡するため、列の追加/削除/名前変更やパーティション定義の変更にデータの再書き込みが不要。
パターン
隠しパーティショニング + MERGE upsert で作成:
CREATE TABLE lake.db.orders (order_id BIGINT, customer_id BIGINT, amount DECIMAL, ordered_at TIMESTAMP)
USING iceberg PARTITIONED BY (days(ordered_at));
MERGE INTO lake.db.orders t USING staging s ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;
メンテナンス（Spark プロシージャ）:
CALL lake.system.rewrite_data_files('db.orders'); -- 小ファイルをコンパクション
CALL lake.system.expire_snapshots('db.orders', TIMESTAMP '2026-08-01 00:00:00');
CALL lake.system.remove_orphan_files(table => 'db.orders');
タイムトラベル — 監査や復元のために以前のスナップショットを読み取る：
SELECT * FROM lake.db.orders VERSION AS OF <snapshot_id>。
パーティション進化 — ALTER TABLE ... ADD PARTITION FIELD bucket(16, customer_id) は新しいデータにのみ適用され、古いデータはそのまま有効。
よくある落とし穴
- ストリーミングテーブルでコンパクションをしない — 小ファイルの爆発により読み取り速度が低下。
- スナップショットを期限切れにしない — メタデータとストレージが無制限に増大するため、タイムトラベルの保持期間内にスケジュールして期限切れにする。
- 過剰なパーティショニング（例：低ボリュームなのに時間単位） — 小さなパーティションが多すぎるため、データサイズに合わせた粒度にすること。
- 明示的な導出パーティション列を追加する — 隠しパーティショニングの目的が失われるため、変換を使用すること。
- MERGE の代わりに無条件で上書きする — 冪等性が失われるため、キーで MERGE を使用すること。
- 孤立ファイルを削除する際の時間枠が短すぎる — 書き込み中のライターが必要とするファイルを削除する可能性があるため、安全なカットオフを使用すること。