🧊

Docker不要！uv + PyIcebergでApache Icebergのメタデータ構造を理解する

2025/12/27に公開

はじめに

Apache Icebergは、データレイク上でデータウェアハウスのような機能を実現するオープンテーブルフォーマットです。AWS S3 Tables、Snowflake、BigQueryなど主要クラウドサービスでも採用が進んでいるようです。

この記事では、「まずはIcebergの仕組みを軽く理解したい」ということを優先して、Docker不要で、uv + PyIcebergだけを使ってIcebergのメタデータ構造を理解することを目指します。

対象読者

Apache Icebergの仕組みを理解したい方
Docker環境なしで軽量にIcebergを試したい方
S3 TablesやAthenaを使う前にIcebergの基礎を押さえたい方
uvを試してみたい方

この記事で学べること

Icebergのメタデータ構造（metadata.json、マニフェストリスト、マニフェストファイル）
データ追加・更新・削除時のファイル変化
スナップショットとタイムトラベルの仕組み

環境構築

uvのインストール

uvはRust製の高速なPythonパッケージマネージャーです。pip + venv + pyenvの機能を統合しており、依存解決も高速です。

# macOS (Homebrew)
brew install uv

# または公式インストーラー
curl -LsSf https://astral.sh/uv/install.sh | sh

インストール確認：

$ which uv
/opt/homebrew/bin/uv  # Homebrewでインストールした場合

プロジェクトの作成

mkdir iceberg-tutorial
cd iceberg-tutorial

# プロジェクト初期化
uv init

# Pythonバージョン指定（必要に応じてuvが自動ダウンロード）
uv python pin 3.12

# PyIcebergと依存パッケージを追加
uv add "pyiceberg[pyarrow,sql-sqlite]"
uv add pandas

環境が正しく構築されたか確認します：

$ uv run python --version
Using CPython 3.12.8 interpreter at: /opt/homebrew/opt/python@3.12/bin/python3.12
Creating virtual environment at: .venv
Python 3.12.8

これでpyproject.tomlとuv.lockが生成され、依存関係が解決されます。

Icebergのメタデータ構造（概要）

本題に入る前に、Icebergのメタデータ構造を簡単に説明します。

┌─────────────────────────────────────┐
│  カタログ層                          │  ← 「現在のメタデータはどこか」を管理
│  (Glue Data Catalog, SQLite等)      │
├─────────────────────────────────────┤
│  メタデータ層                        │  ← スナップショット、スキーマ情報
│  (metadata.json, manifest list/file)│
├─────────────────────────────────────┤
│  データ層                            │  ← 実際のParquetファイル
│  (Parquetファイル群)                 │
└─────────────────────────────────────┘

AWSでいえば、カタログ層がGlue Data Catalog、データ層がS3に対応します。今回はローカル環境でSQLiteをカタログとして使用します。

ハンズオン

ここからはPython REPLで一つずつ確認しながら進めます。

uv run python

Step 1: インポートとカタログ設定

from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import StringType, LongType, NestedField
from pathlib import Path

# warehouseディレクトリの準備
warehouse_path = Path("./warehouse").absolute()
warehouse_path.mkdir(exist_ok=True)
print(f"Warehouse: {warehouse_path}")

Warehouse: /Users/yourname/iceberg-tutorial/warehouse

Step 2: カタログの作成

catalog = load_catalog(
    "local",
    **{
        "type": "sql",
        "uri": "sqlite:///iceberg_catalog.db",
        "warehouse": f"file://{warehouse_path}"
    }
)

print(catalog)

local (<class 'pyiceberg.catalog.sql.SqlCatalog'>)

この時点で、プロジェクト直下にiceberg_catalog.db（SQLiteファイル）が作成されます。これがカタログとして「どのテーブルがどこにあるか」を管理します。

Step 3: ネームスペースの作成

# 既存のネームスペース確認
print(catalog.list_namespaces())

# ネームスペース作成
catalog.create_namespace("tutorial")

# 作成後の確認
print(catalog.list_namespaces())

[]
[('tutorial',)]

Step 4: スキーマ定義

schema = Schema(
    NestedField(1, "user_id", LongType(), required=True),
    NestedField(2, "name", StringType(), required=True),
    NestedField(3, "email", StringType(), required=False),
)

# スキーマの中身を確認
for field in schema.fields:
    print(f"  {field.field_id}: {field.name} ({field.field_type}) required={field.required}")

  1: user_id (long) required=True
  2: name (string) required=True
  3: email (string) required=False

required=TrueのフィールドはNULLを許容しません。これは後のデータ追加時に重要になります。

Step 5: テーブル作成

table = catalog.create_table("tutorial.users", schema=schema)
print(table)

users(
  1: user_id: required long,
  2: name: required string,
  3: email: optional string
)

Step 6: メタデータの確認

metadata = table.metadata

print(f"テーブルUUID: {metadata.table_uuid}")
print(f"フォーマットバージョン: {metadata.format_version}")
print(f"ロケーション: {metadata.location}")

テーブルUUID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
フォーマットバージョン: 2
ロケーション: file:///Users/yourname/iceberg-tutorial/warehouse/tutorial/users

Step 7: ファイル構造の確認

この時点で別ターミナルからwarehouseディレクトリを確認してみましょう。

$ tree warehouse
warehouse
└── tutorial
    └── users
        └── metadata
            └── 00000-xxxxxxxx.metadata.json

テーブル作成直後は、metadataディレクトリに最初のmetadata.jsonだけが存在します。まだデータを追加していないのでdataディレクトリはありません。

Step 8: metadata.jsonの中身を確認

import json

metadata_dir = warehouse_path / "tutorial" / "users" / "metadata"
metadata_files = sorted(metadata_dir.glob("*.metadata.json"))
print(f"メタデータファイル: {[f.name for f in metadata_files]}")

# metadata.jsonを読む
with open(metadata_files[-1]) as f:
    meta = json.load(f)

print(json.dumps(meta, indent=2)[:500])  # 長いので先頭部分だけ

metadata.jsonにはスキーマ定義、パーティション仕様、スナップショット履歴などが記録されています。

データの追加

Step 9: PyArrowでデータを準備して追加

PyIcebergでデータを追加する際は、PyArrowのTableを使用します。

import pyarrow as pa

# スキーマを明示的に定義（required/optionalを合わせる）
arrow_schema = pa.schema([
    pa.field("user_id", pa.int64(), nullable=False),
    pa.field("name", pa.string(), nullable=False),
    pa.field("email", pa.string(), nullable=True),
])

data = pa.table({
    "user_id": [1, 2, 3],
    "name": ["Alice", "Bob", "Charlie"],
    "email": ["alice@example.com", "bob@example.com", None],
}, schema=arrow_schema)

table.append(data)
print("3件追加しました")

PyArrowでテーブルを作成する際、スキーマを明示的に指定しないと全カラムがnullable=True（optional）になります。Icebergテーブルのスキーマでrequired=Trueとしたカラムがある場合、スキーマ不一致エラーが発生します。

ValueError: Mismatch in fields:
┃    ┃ Table field               ┃ Dataframe field           ┃
│ ❌ │ 1: user_id: required long │ 1: user_id: optional long │

このエラーが出た場合は、pa.schema()でスキーマを明示的に定義してください。

Step 10: スナップショットの確認

table.refresh()  # メタデータ再読み込み

for snapshot in table.metadata.snapshots:
    print(f"Snapshot ID: {snapshot.snapshot_id}")
    print(f"  Operation: {snapshot.summary.get('operation')}")
    print(f"  Added records: {snapshot.summary.get('added-records')}")

Snapshot ID: 5815774430499421792
  Operation: Operation.APPEND
  Added records: 3

Step 11: 追加データでスナップショットを増やす

more_data = pa.table({
    "user_id": [4, 5],
    "name": ["David", "Eve"],
    "email": ["david@example.com", "eve@example.com"],
}, schema=arrow_schema)

table.append(more_data)
table.refresh()

print(f"スナップショット数: {len(table.metadata.snapshots)}")

スナップショット数: 2

Step 12: ファイル構造の変化を確認

別ターミナルでwarehouseディレクトリを確認すると、データ追加によってファイルが増えていることがわかります。

$ tree warehouse
warehouse
└── tutorial
    └── users
        ├── data
        │   ├── 00000-0-xxxxxxxx.parquet
        │   └── 00000-0-yyyyyyyy.parquet
        └── metadata
            ├── 00000-xxxxxxxx.metadata.json
            ├── 00001-xxxxxxxx.metadata.json
            ├── 00002-xxxxxxxx.metadata.json
            ├── xxxxxxxx-m0.avro
            ├── yyyyyyyy-m0.avro
            ├── snap-xxxx-0-xxxxxxxx.avro
            └── snap-yyyy-0-yyyyyyyy.avro

ファイルの役割

Step 12で確認したファイル構造について、それぞれの役割を説明します。

ファイル種別	説明
`*.parquet`	実データ（カラムナフォーマット）
`*.metadata.json`	テーブル状態のスナップショット。操作ごとに新規作成
`snap-*.avro`	マニフェストリスト。スナップショットに含まれるマニフェストファイルの一覧
`*-m0.avro`	マニフェストファイル。Parquetファイルの一覧と統計情報

階層構造：

metadata.json（現在の状態）
    ↓ 参照
snap-*.avro（マニフェストリスト）
    ↓ 参照
*-m0.avro（マニフェストファイル）
    ↓ 参照
data/*.parquet（実データ）

レコードの更新

Step 13: 現在のデータを確認

df = table.scan().to_pandas()
print(df)

   user_id     name              email
0        1    Alice  alice@example.com
1        2      Bob    bob@example.com
2        3  Charlie               None
3        4    David  david@example.com
4        5      Eve    eve@example.com

Step 14: レコードの更新

PyIcebergではoverwriteを使って条件に合うデータを置き換えます。

# user_id=2 (Bob) のemailを更新
updated_data = pa.table({
    "user_id": [2],
    "name": ["Bob"],
    "email": ["bob.new@example.com"],
}, schema=arrow_schema)

table.overwrite(updated_data, overwrite_filter="user_id == 2")
print("更新完了")

Step 15: 更新後のデータとメタデータを確認

table.refresh()
df = table.scan().to_pandas()
print(df)

   user_id     name                email
0        2      Bob  bob.new@example.com
1        1    Alice    alice@example.com
2        3  Charlie                 None
3        4    David    david@example.com
4        5      Eve      eve@example.com

Bobのemailが更新されています。

for snap in table.metadata.snapshots:
    print(f"Snapshot ID: {snap.snapshot_id}")
    print(f"  Operation: {snap.summary.get('operation')}")
    print(f"  Added files: {snap.summary.get('added-data-files', 0)}")
    print(f"  Deleted files: {snap.summary.get('deleted-data-files', 0)}")
    print()

Snapshot ID: 5815774430499421792
  Operation: Operation.APPEND
  Added files: 1
  Deleted files: None

Snapshot ID: 3879896634347338753
  Operation: Operation.APPEND
  Added files: 1
  Deleted files: None

Snapshot ID: 5929255525955454062
  Operation: Operation.OVERWRITE
  Added files: 1
  Deleted files: 1

Snapshot ID: 7234342154326247462
  Operation: Operation.APPEND
  Added files: 1
  Deleted files: None

overwrite操作では、対象データを含むファイルが論理削除（Deleted files: 1）され、新しいファイルが追加（Added files: 1）されます。これがCopy-on-Write方式です。

レコードの削除

Step 16: レコードの削除

PyIcebergではdeleteメソッドで条件に合うレコードを削除できます。

# user_id=3 (Charlie) を削除
table.delete(delete_filter="user_id == 3")
print("削除完了")

Step 17: 削除後のデータとメタデータを確認

table.refresh()
df = table.scan().to_pandas()
print(df)

   user_id   name                email
0        1  Alice    alice@example.com
1        2    Bob  bob.new@example.com
2        4  David    david@example.com
3        5    Eve      eve@example.com

Charlieが削除されています。

for snap in table.metadata.snapshots:
    print(f"Snapshot ID: {snap.snapshot_id}")
    print(f"  Operation: {snap.summary.get('operation')}")
    print(f"  Added files: {snap.summary.get('added-data-files', 0)}")
    print(f"  Deleted files: {snap.summary.get('deleted-data-files', 0)}")
    print(f"  Deleted records: {snap.summary.get('deleted-records', 0)}")
    print()

Snapshot ID: 5815774430499421792
  Operation: Operation.APPEND
  Added files: 1
  Deleted files: None
  Deleted records: None

Snapshot ID: 3879896634347338753
  Operation: Operation.APPEND
  Added files: 1
  Deleted files: None
  Deleted records: None

Snapshot ID: 5929255525955454062
  Operation: Operation.OVERWRITE
  Added files: 1
  Deleted files: 1
  Deleted records: 3

Snapshot ID: 7234342154326247462
  Operation: Operation.APPEND
  Added files: 1
  Deleted files: None
  Deleted records: None

Snapshot ID: 4494934084334135847
  Operation: Operation.OVERWRITE
  Added files: 1
  Deleted files: 1
  Deleted records: 2

削除操作はOperation.OVERWRITEとして記録され、deleted-recordsとdeleted-data-filesが増加しています。Icebergでは削除も内部的にはファイルの書き換え（Copy-on-Write）として処理されます。

3番目のスナップショット（Deleted records: 3）はStep 14での更新時、5番目のスナップショット（Deleted records: 2）は今回の削除時のものです。

タイムトラベル

Icebergの大きな特徴の一つが、過去のスナップショットにアクセスできる「タイムトラベル」機能です。

Step 18: 過去のスナップショットでデータを読む

# 最初のスナップショットのIDを取得
first_snapshot_id = table.metadata.snapshots[0].snapshot_id

# 過去のスナップショットでスキャン
old_df = table.scan(snapshot_id=first_snapshot_id).to_pandas()
print("=== 最初のappend時点のデータ ===")
print(old_df)

=== 最初のappend時点のデータ ===
   user_id     name              email
0        1    Alice  alice@example.com
1        2      Bob    bob@example.com
2        3  Charlie               None

更新前のBobのメールアドレスや、削除したCharlieのレコードが確認できます。

まとめ

この記事では、uv + PyIcebergを使ってApache Icebergのメタデータ構造を確認しました。

学んだこと

カタログ: テーブルの場所を管理（今回はSQLite、本番ではGlue Data Catalogなど）
metadata.json: テーブルの状態を記録。操作ごとに新しいファイルが作成される
マニフェストリスト/ファイル: データファイルの一覧と統計情報を階層的に管理
スナップショット: 各時点のテーブル状態。タイムトラベルの基盤
Copy-on-Write: 更新・削除時は元ファイルを残して新ファイルを作成

次のステップ

AWS S3 Tables: Icebergをマネージドで使えるAWSサービス
dbt-athena: dbtからAthena経由でIcebergテーブルを操作
パーティショニング: Hidden Partitioningによる柔軟なパーティション管理

Icebergの内部構造を理解しておくと、S3 TablesやAthenaを使う際にも「裏で何が起きているか」がわかり、トラブルシューティングや設計判断に役立つのではないかと思います。

参考資料

GitHubで編集を提案