実際の問題：ETLパイプライン実行後、元データが消える
こんな経験をしたことがある。ETLパイプラインが午前2時に動き出し、salesデータをtransformして既存のParquetファイルを上書きした。翌朝、analyticsチームから先月のレポート結果が突然変わったとクレームが来た。ログを調べても原因がわからない——元データが完全に上書きされていたからだ。
これは珍しいケースではない。中規模のパイプラインでも、jobがParquetを上書きするたびにスナップショットの履歴が完全に失われる——事前に手動バックアップしない限り、ロールバックする方法がない。多くのチームは日付でファイル名をつけて対応している：sales_2024_11_01.parquet、sales_2024_11_02.parquet……だがこの方法はストレージを食い、クエリが複雑になり、それでもACID保証がない。
なぜ純粋なParquetでは不十分なのか？
Parquetはanalyticsワークロードに非常に優れている——カラム型ストレージ、高い圧縮率、高速な読み取り。しかし、Parquetは純粋なファイルフォーマットであり、transactionやversioningという概念がない。Parquetファイルを上書きすると：
- 変更履歴がない（audit trail）
- 処理途中でjobが失敗してもrollbackできない
- Schema evolutionが難しい——カラムの追加・削除でdownstreamのパイプラインが壊れやすい
- 複数jobによる同時書き込みがrace conditionやdata corruptionを引き起こす
Parquetが悪いわけではない——設計された目的を正しく果たしている。足りないのは、その上に乗るmetadata管理とtransactionの層だ。
Apache Iceberg + DuckDBはどう解決するか
Icebergはまさにそのギャップを埋めるために生まれた。内部はParquetのままだが、Icebergはその上に4つのコア機能を持つmetadata管理レイヤーを追加する：
- Snapshot-based versioning：書き込みのたびに新しいsnapshotを作成し、古いsnapshotは削除しない
- ACID transactions：同時書き込みによるデータ破損を心配しなくてよい
- Schema evolution：データ全体を書き直さずにカラムの追加・削除・リネームが可能
- Time Travel：過去の任意の時点のデータをクエリできる
DuckDBはquery engineとして機能する：軽量で高速、ラップトップ上ですぐに動く。バージョン0.10+から、extension icebergがIceberg tablesの直接読み取りを公式サポートしている。スタック全体がローカル——AWSもHive Metastoreも不要だ。
環境セットアップ
Python 3.9+さえあれば十分で、他のサービスへの依存はない。
# virtual environmentを作成
python -m venv iceberg-env
source iceberg-env/bin/activate
# PyIcebergとDuckDBをインストール
pip install "pyiceberg[pyarrow,duckdb]" duckdb pandas pyarrowインストール後にバージョンを確認：
import pyiceberg
import duckdb
print(pyiceberg.__version__) # >= 0.7.0
print(duckdb.__version__) # >= 0.10.0はじめてのIceberg CatalogとTableを作成する
PyIcebergはmetadataを管理するためにcatalogが必要だ。ローカル環境では、SqlCatalogを使ってmetadataをSQLiteに保存する——追加のサービスセットアップは一切不要だ。
from pyiceberg.catalog.sql import SqlCatalog
import pyarrow as pa
import os
os.makedirs("warehouse/data", exist_ok=True)
catalog = SqlCatalog(
"local",
**{
"uri": "sqlite:///warehouse/catalog.db",
"warehouse": "file://warehouse",
}
)
catalog.create_namespace("analytics")
schema = pa.schema([
pa.field("order_id", pa.int64()),
pa.field("customer_id", pa.int64()),
pa.field("amount", pa.float64()),
pa.field("status", pa.string()),
pa.field("created_at", pa.timestamp("us")),
])
table = catalog.create_table("analytics.orders", schema=schema)
print("Table created:", table.identifier)データを書き込みSnapshotを作成する
import pandas as pd
from datetime import datetime, timedelta
# Batch 1: 昨日のデータ
df1 = pd.DataFrame({
"order_id": [1, 2, 3],
"customer_id": [101, 102, 103],
"amount": [150.0, 230.5, 89.99],
"status": ["completed", "pending", "completed"],
"created_at": [datetime.now() - timedelta(days=1)] * 3
})
table.append(pa.Table.from_pandas(df1))
snapshot_1 = table.current_snapshot().snapshot_id
print(f"Snapshot 1 ID: {snapshot_1}")
# Batch 2: 新しい注文を追加
df2 = pd.DataFrame({
"order_id": [4, 5],
"customer_id": [104, 105],
"amount": [320.0, 45.0],
"status": ["completed", "refunded"],
"created_at": [datetime.now()] * 2
})
table.append(pa.Table.from_pandas(df2))
snapshot_2 = table.current_snapshot().snapshot_id
print(f"Snapshot 2 ID: {snapshot_2}")この時点でIcebergは2つのsnapshotを保持している。snapshot 1の元データはディスク上にそのまま残り、削除も上書きもされない——これこそが純粋なParquetにはできないことだ。
DuckDBによるTime Travel Query
いよいよ最も価値ある部分だ。DuckDBはIceberg tableを直接読み取れる。特定のsnapshotを指定したクエリも同様だ：
import duckdb
conn = duckdb.connect()
conn.execute("INSTALL iceberg; LOAD iceberg;")
# 現在の状態をクエリ — 5件の注文すべて
result = conn.execute("""
SELECT * FROM iceberg_scan('warehouse/analytics/orders')
ORDER BY order_id
""").fetchdf()
print("Current state (5 rows):")
print(result)
# Time Travel — snapshot 1をクエリ（最初の3件のみ）
result_v1 = conn.execute(f"""
SELECT * FROM iceberg_scan(
'warehouse/analytics/orders',
snapshot_id = {snapshot_1}
)
ORDER BY order_id
""").fetchdf()
print(f"\nSnapshot {snapshot_1} — 3 rows only:")
print(result_v1)午前2時のパイプラインが誤って上書きしても、直前のsnapshotをクエリすれば元データがすぐに取れる——バックアップのリストアも、ダウンタイムも不要だ。
Schema Evolution — 既存のパイプラインを壊さずカラムを追加する
from pyiceberg.types import FloatType
# schemaにdiscount_pctカラムを追加
with table.update_schema() as update:
update.add_column("discount_pct", FloatType())
# 古いsnapshotは引き続き読み取れる — 新しいカラムは古いデータにNULLを返す
result = conn.execute("""
SELECT order_id, amount, discount_pct
FROM iceberg_scan('warehouse/analytics/orders')
""").fetchdf()
print(result)
# discount_pct = NaN（5件の古い注文）— 完全に正常な動作純粋なParquetでは、このようにカラムを追加すると古いsnapshotのreaderがschemaの不一致で壊れることが多い。Icebergは自動で処理する——新しいカラムは古いデータにNULLを返し、古いreaderは壊れない。
実戦から学んだTipsとBest Practices
1. 最初からsnapshotのretention policyを設定する
Snapshotは時間とともに積み重なり、ディスクを消費する。定期的にexpireさせるべきだ：
from datetime import datetime, timedelta
# 直近7日間のsnapshotを保持する
table.expire_snapshots(
older_than_ms=int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
).commit()
print("Expired old snapshots")2. クエリを高速化するために時間でテーブルをpartitionする
table作成時にpartitionを宣言する（Iceberg v1では後から追加できない）：
from pyiceberg.partitioning import PartitionSpec, PartitionField
from pyiceberg.transforms import DayTransform
# source_id=5はschema内のcreated_atカラムのインデックス
spec = PartitionSpec(
PartitionField(
source_id=5,
field_id=1000,
transform=DayTransform(),
name="created_day"
)
)
# table作成時に宣言する
table = catalog.create_table(
"analytics.orders_partitioned",
schema=schema,
partition_spec=spec
)DuckDBはファイル全体を読み込む代わりに、必要なpartitionだけをスキャンする。100GBのテーブルでは、特定の1日分のクエリがfull scanより20〜50倍速くなることもある。
3. ingest前にソースデータを正規化する
複数のソースからデータを受け取る際、Icebergに投入する前にCSVをJSONに変換して正規化する必要があることがよくある。10MB未満の小さなファイルであれば、ブラウザ上のコンバーターをtoolcraft.app/ja/tools/data/csv-to-jsonで使っている——クライアントサイドで動くため機密データが漏れる心配がなく、ソースごとにone-offスクリプトを書くより遥かに便利だ。
4. 複雑なtransformにはdbtと組み合わせる
DuckDBにはdbt用の公式adapter（dbt-duckdb）がある。スタック全体が完全ローカル：Iceberg storage + DuckDB compute + dbt transforms = クラウドコストゼロのmini data warehouse。小規模チームやproductionにスケールアップする前のproof-of-conceptに最適だ。
始めるための最善のアプローチ
Parquetファイルでdata pipelineを管理していて、データを失ったり履歴を追えなかった経験がある人に、Icebergへの段階的な移行を勧める：
- 最もcriticalでないテーブルから始め、dev環境でTime Travelをテストする
- ローカル環境では
SqlCatalogとSQLiteを使う——複雑なinfraは不要 - 最初から
expire_snapshotsポリシーを設定し、後回しにしない - 慣れてきたら、必要に応じてREST CatalogやAWS Glue上のproductionへmigrationする
Icebergはあらゆる用途に適しているわけではない——数百万rows以下の小さなテーブルでは、metadata管理のオーバーヘッドに見合わない。しかし、データが数十GBに達したとき、またはaudit trailが必須のとき（金融、コンプライアンス）、これは自分が使ってきた中で最善の選択肢だ。このstackに移行してから、パイプラインのデバッグで夜を明かすことがなくなった。
