レイクハウスのテーブル形式 — Apache Iceberg をメタデータで理解する

Spark と pyiceberg で一つのカタログを共有し、名前を変え、消したテーブルを復活させる

目標

JDBCカタログ(SQLiteファイル1つ)をSparkとpyicebergが一緒に使うようにし、コミットがカタログのポインター1つを書き換える作業であることを、コミット前後の値で確認します。テーブル名の変更と削除がカタログでしか起きず、ファイルはそのままであることを見て、1つのmetadataファイルで削除したテーブルを復元します。

なぜ重要なのか

Icebergテーブルの「本当の状態」はmetadataファイルにあり、カタログは名前1つをそのファイル1つにつなぐ小さなテーブルです。ところが、この小さなテーブルが同時書き込みの裁定役です。2つのライターが同時にコミットすると、カタログは「自分が読んだポインターがまだそのままのときだけ書き換える」という条件で、一方だけを受け付けます。カタログがこのアトミックな置換をできなければ、テーブルが壊れます。同じカタログを複数のエンジンが見ていれば、Sparkが作ったテーブルをPythonのジョブが引き継いで書き、その結果をまたSparkが読みます。一方で、名前・場所・ファイルが別々の層であることを知らないと、事故が起きます。名前を変えればファイルも移動したと思い込んで古いパスを消したり、DROP TABLEで容量が空いたと思い込んで待ち続けたり、逆に誤って消したテーブルを永久に失ったと諦めたりします。

ステップ

/root/ice/cat/make.py(アプリice-cat-make)でlake.cat.orders(format-version 2)を作成し、2026-03-01・03-02・03-03の3つのファイルを1回でコミットしてください。
/root/ice/cat/list.py(pyiceberg、load_catalog("lake"))で、ネームスペースとテーブルの一覧を、/root/ice/cat/out/tables.jsonに書いてください。
/root/ice/cat/customers.py(pyiceberg)で/data/ice/customers.csvを読み込み、lake.cat.customersを作成して書き込んでください。
/root/ice/cat/join.py(アプリice-cat-join)で2つのテーブルをcustomer_idで結合し、等級別の注文数のテーブルlake.cat.tier_counts(tier, orders)を作成してください。
lake.cat.ordersに2026-03-04をコミットしながら、コミット前のパス、コミット後のパス、コミット後のprevious列の値を、/root/ice/cat/out/pointer.jsonに書いてください。
lake.cat.tier_countsをlake.cat.tier_summaryに名前変更し、変更前後のmetadataパスを、/root/ice/cat/out/rename.jsonに書いてください。
lake.cat.tier_summaryをPURGEなしで削除したあと、残ったデータファイル数を数え、削除直前のmetadataファイルでlake.cat.tier_restoredを登録して、/root/ice/cat/out/restore.jsonに書いてください。
/root/ice/cat/report.mdに、## 포인터、## 이름과 위치、## 지우기와 되살리기の3つの節を書いてください(3つの見出しは順に、韓国語で「ポインター」「名前と場所」「削除と復元」を意味する語句です)。

参考

カタログはsqlite3 /root/ice/catalog.db "select * from iceberg_tables"でそのまま見られます。ice-loc <ns>.<표>(プレースホルダーはネームスペースとテーブル名です)は、現在のmetadataのパスを出力します。
pyicebergの設定は~/.pyiceberg.yamlにあります(カタログ名lake、type: sql)。Sparkの設定は/opt/spark/conf/spark-defaults.confです。
pyicebergが初めてカタログを開くとき、「v0 schema」の警告が出力されます。JavaのJdbcCatalogが作った古い形(ビュー用の列がない)という意味で、テーブルへの書き込みには問題ありません。
RENAME TOの後ろの新しい名前には、カタログを付けません(cat.tier_summary)。lake.cat.tier_summaryと書くと、Sparkがlake.catというネームスペースを探しているうちに、NoSuchNamespaceExceptionで止まります(実測)。
よくある間違いは、ステップ7でDROP TABLE … PURGEを使うことです。ファイルまで削除されて、復元できなくなります。そうなってしまったら、ステップ4からやり直してください。
公式ドキュメント: JDBC Catalog・Spark DDL・Spark Procedures — register_table・pyiceberg — SQL Catalog

Sparkがテーブルを作成する: コミット1回

/root/ice/cat/make.pyをアプリ名ice-cat-makeで作成し、lake.catネームスペースとlake.cat.orders(列は6つ、'format-version' = '2')を作って、2026-03-01・03-02・03-03の3つのファイルを1回のappend()で書き込んでください。

spark.read.csv([경로1, 경로2, 경로3], …)(プレースホルダーは3つのパスです)のようにリストを渡すと、1つのDataFrameになります。コミットが1回なら、スナップショットも1つです。採点ツールは、最初のスナップショットの行数が3つのファイルの合計であるか、要約にSparkが残したengine-nameがあるかを確認します。

pyicebergで同じカタログを見る

/root/ice/cat/list.pyをpyicebergのload_catalog("lake")で作成し、ネームスペースとテーブルの一覧を、/root/ice/cat/out/tables.jsonに{"namespaces": ["cat", …], "tables": ["cat.orders", …]}の形で書くようにして、python3 list.pyで実行してください。

pyicebergは、~/.pyiceberg.yamlからlakeカタログの設定(type: sql、uri: sqlite:////root/ice/catalog.db)を読みます。サーバーなしで同じファイルを開くので、Sparkが作ったテーブルがそのまま見えます。名前はタプルで出てくるので、ドットでつないでください。

pyicebergがテーブルを作成する

/root/ice/cat/customers.pyで/data/ice/customers.csvをpyarrowで読み込み(signup_dateはdate32)、lake.cat.customersを作成して書き込んでください。もう一度実行しても行が2倍にならないように、overwrite()を使ってください。

create_table_if_not_exists("cat.customers", schema=arrow_table.schema)は、pyarrowのスキーマをIcebergのスキーマに変換し、列ごとにフィールドIDを付けます。pyicebergが書いたスナップショットの要約には、Sparkが残すengine-nameがありません。採点ツールは、それで誰が書いたかを判別します。

Sparkが2つのエンジンのテーブルを結合する

/root/ice/cat/join.pyをアプリ名ice-cat-joinで作成し、lake.cat.ordersとlake.cat.customersをcustomer_idで結合して、等級(tier)別の注文数をlake.cat.tier_counts(列はtier、orders)として作成してください(CREATE TABLE … AS SELECT)。

誰が書いたかにかかわらず、Icebergテーブルは同じ仕様のmetadataとParquetなので、エンジンは関係ありません。採点ツールは、元のCSVから等級別の注文数を直接計算して比較します(このとき、ordersは3月1–3日分です)。

コミット1回でポインターが1つ動く

lake.cat.ordersの現在のmetadataパスを書き留めてから、/root/ice/cat/append.py(日付の引数)で2026-03-04をコミットし、コミット後のパスとカタログのprevious_metadata_locationを読み取り、/root/ice/cat/out/pointer.jsonに{"before", "after", "previous_after"}の形で書いてください。

コミットは、新しいmetadataファイルを先にすべて書いておいてから、カタログで「現在の値がbeforeのときだけafterに書き換える」という条件付き更新を1回行って終わります。そのため、コミット後のprevious列は、コミット前のパスと同じでなければなりません。jq -n --argで、シェル変数をJSONにまとめられます。

名前はカタログにしかない

ALTER TABLE lake.cat.tier_counts RENAME TO cat.tier_summaryを実行して(新しい名前にはカタログlakeを付けません)、変更前(tier_counts)と変更後(tier_summary)のmetadataパスを、/root/ice/cat/out/rename.jsonに{"before", "after"}の形で書いてください。

JDBCカタログの名前変更は、iceberg_tablesの1行のtable_nameを書き換えるUPDATEです。テーブルの場所(location)とファイルはそのままなので、新しい名前のテーブルは、今も…/cat/tier_counts/の下のファイルを指しています。採点ツールは、2つのパスが同じであるか、古い名前がカタログから消えているかを確認します。

削除したテーブルを1つのmetadataファイルで復元する

lake.cat.tier_summaryのmetadataパスを書き留めてから、PURGEなしでDROP TABLEを実行し、/root/ice/warehouse/cat/tier_counts/dataに残ったParquetファイル数を数えて、CALL lake.system.register_table(table => 'lake.cat.tier_restored', metadata_file => '<그 경로>')(プレースホルダーは、直前に書き留めたmetadataのパスです)で復元してください。/root/ice/cat/out/restore.jsonには、{"metadata_file", "files_left"}を書いてください。

PURGEなしのDROPは、カタログの1行だけを消します。metadataファイルがスキーマ・スナップショット・ファイル一覧をすべて持っているので、そのパスさえわかれば、どのカタログにでも再登録できます(カタログを移すときや、バックアップから復元するときに使う方法です)。PURGEを付けたなら、ファイルが削除されていて復元できません。

カタログがすることと、しないこと

/root/ice/cat/report.mdに、## 포인터、## 이름과 위치、## 지우기와 되살리기の3つの節を書いてください(3つの見出しは順に、韓国語で「ポインター」「名前と場所」「削除と復元」を意味する語句です)。3つ目の節には、ステップ7で数えた残りのファイル数を、数字で入れてください。

本番でカタログを切り替えるとき(例: JDBC → REST)や、バックアップからテーブルを復元するときに、何を移す必要があり、何をそのままにしておけばよいかを書いてみてください。