R2

R2 シンクは、パイプラインで処理したデータを生ファイルとして R2 オブジェクトストレージ へ書き込みます。現在、JSON と Parquet 形式への書き込みに対応しています。

R2 シンクを作成するには、pipelines sinks create コマンドを実行し、シンクの種類と対象の バケット を指定します。

npx wrangler pipelines sinks create my-sink \
  --type r2 \
  --bucket my-bucket

R2 シンクは、次の 2 つの出力形式に対応しています。

改行区切りの JSON ファイルとしてデータを書き込みます。JSON シンクは次の 2 つの圧縮モードに対応しています。uncompressed（デフォルト）と gzip です。

--format json --compression gzip

クエリ性能と圧縮率を高めるため、Parquet ファイルとしてデータを書き込みます。

--format parquet --compression zstd

Parquet の圧縮オプション:

• zstd（デフォルト）- 圧縮率が最も高い
• snappy - 圧縮が最も速い
• gzip - 圧縮率が高く、広くサポートされている
• lz4 - 高速な圧縮で、圧縮率も妥当
• uncompressed - 圧縮なし

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

--target-row-group-size 256

ファイルは、パーティション分割されたディレクトリ構造の中に UUID 名で書き込まれます。たとえば、パスが analytics でデフォルトのパーティション分割の場合は次のとおりです。

analytics/year=2025/month=09/day=18/002507a5-d449-48e8-a484-b1bea916102f.parquet

ファイルを書き込むバケット内のベースディレクトリを設定します。

--path analytics/events

R2 シンクは、設定可能なパターンで時刻ごとにファイルを自動でパーティション分割します。デフォルトのパターンは year=%Y/month=%m/day=%d（Hive 形式のパーティション分割）です。

--partitioning "year=%Y/month=%m/day=%d/hour=%H"

使える書式指定子は、strftime のドキュメント ↗ を参照してください。

R2 へファイルを書き込むタイミングを制御します。要件に応じて設定してください。

• 小さい値: 書き込み頻度が高く、ファイルは小さく、レイテンシは低い
• 大きい値: 書き込み頻度が低く、ファイルは大きく、クエリ性能は高い

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

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

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

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

R2 シンクには、バケットへデータを書き込むための API 認証情報（Access Key ID と Secret Access Key）が必要です。Object Read & Write 権限 が必要です。

npx wrangler pipelines sinks create my-sink \
  --type r2 \
  --bucket my-bucket \
  --access-key-id YOUR_ACCESS_KEY_ID \
  --secret-access-key YOUR_SECRET_ACCESS_KEY