R2 Data Catalog

R2 Data Catalog sink は、パイプラインで処理したデータを Apache Iceberg ↗ テーブルとして R2 Data Catalog に書き込みます。Iceberg テーブルは、分析ワークロード向けに ACID トランザクション、スキーマ進化、タイムトラベルを提供します。

R2 Data Catalog sink を作成するには、pipelines sinks create コマンドを実行し、sink の種類、対象バケット、名前空間、テーブル名を指定します。

npx wrangler pipelines sinks create my-sink \
  --type r2-data-catalog \
  --bucket my-bucket \
  --namespace my_namespace \
  --table my_table \
  --catalog-token YOUR_CATALOG_TOKEN

指定した名前空間とテーブルがなければ、sink が作成します。既存の Iceberg テーブル向けには sink を作成できません。

R2 Data Catalog sink が対応するのは Parquet フォーマットだけです。Iceberg テーブルでは JSON フォーマットは使えません。

ストレージとクエリ性能に合わせて、Parquet の圧縮を設定します。

--compression zstd

利用できる圧縮オプション:

• zstd（デフォルト） - 圧縮率がもっとも高い
• snappy - 圧縮がもっとも速い
• gzip - 圧縮率が高く、広く対応されている
• lz4 - 速く、圧縮率も妥当
• uncompressed - 圧縮なし

行グループ ↗ は、Parquet ファイル内でまとめて保存する行の集合です。メモリ使用量とクエリ性能に影響します。目標の行グループサイズを MB 単位で設定します。

--target-row-group-size 256

Iceberg テーブルへの書き込みタイミングを制御します。用途に合わせて設定します。

• 小さい値: 書き込みが頻繁、ファイルが小さい、レイテンシが低い
• 大きい値: 書き込みが少ない、ファイルが大きい、クエリ性能が良い

ファイルを書き込む間隔を設定します（デフォルト: 300 秒、最小: 60 秒）。

--roll-interval 60  # Write files every 60 seconds

R2 Data Catalog sink の最小間隔は 60 秒です。コンパクションの不具合を防ぐためです。Iceberg テーブルは、小さなファイルを大きなファイルへまとめる定期コンパクションが必要です。書き込みが多すぎると、コンパクション処理とマージが衝突します。

新しいファイルを作るまでの最大ファイルサイズを MB 単位で設定します。

--roll-size 100  # Create new file after 100MB

R2 Data Catalog sink には、R2 Admin Read & Write 権限 を持つ API トークンが必要です。この権限で、sink は R2 Data Catalog と R2 ストレージの両方にアクセスできます。

--catalog-token YOUR_CATALOG_TOKEN