はじめに

このガイドでは、次の手順を説明します。

パイプラインがデータカタログへ認証するために必要な API トークンを作成する
簡単な e コマーススキーマで、R2 Data Catalog が管理する Apache Iceberg ↗ テーブルへ書き込む最初のパイプラインを作成する
HTTP エンドポイント経由でサンプルの e コマースデータを送信する
バケット内のデータを確認し、R2 SQL でクエリする

前提条件

Cloudflare アカウント ↗ に登録します。
Node.js ↗ をインストールします。

Node.js のバージョンマネージャー

権限の問題を避け、Node.js のバージョンを切り替えられるよう、Volta ↗ や nvm ↗ などの Node バージョンマネージャーを使います。このガイドの後半で説明する Wrangler には、Node バージョン 16.17.0 以降が必要です。

1. API トークンを作成する

Pipelines は、カタログと R2 の権限を持つ R2 API トークンで R2 Data Catalog に認証する必要があります。

Cloudflare ダッシュボードで R2 object storage ページを開きます。
Overview を開く ↗
Manage API tokens を選びます。
Create Account API token を選びます。
API トークンに名前を付けます。
Permissions で Admin Read & Write 権限を選びます。
Create Account API Token を選びます。
Token value を控えます。

2. 最初のパイプラインを作成する

まず、e コマースのデータ構造を定義するスキーマファイルを作成します。

schema.json を作成します。

{
	"fields": [
		{
			"name": "user_id",
			"type": "string",
			"required": true
		},
		{
			"name": "event_type",
			"type": "string",
			"required": true
		},
		{
			"name": "product_id",
			"type": "string",
			"required": false
		},
		{
			"name": "amount",
			"type": "float64",
			"required": false
		}
	]
}

対話型セットアップで、R2 Data Catalog へ書き込むパイプラインを作成します。

npx wrangler pipelines setup

プロンプトに従います。

Pipeline name: ecommerce を入力します
Stream configuration:
- Enable HTTP endpoint: yes
- Require authentication: no（簡略化のため）
- Configure custom CORS origins: no
- Schema definition: Load from file
- Schema file path: schema.json（またはファイルパス）
Sink configuration:
- Destination type: Data Catalog (Iceberg)
- Setup mode: Simple (recommended defaults)
- R2 bucket name: pipelines-tutorial（存在しない場合は自動作成）
- Table name: ecommerce
- Catalog API token: 手順 1 のトークンを入力します
Review: 概要を確認し、Create resources を選びます
SQL transformation: Simple ingestion (SELECT * FROM stream) を選びます

詳細モードのオプション

シンク設定で Simple ではなく Advanced を選ぶと、次の追加オプションをカスタマイズできます。

Format: 出力ファイル形式（例: Parquet）
Compression: 圧縮アルゴリズム（例: zstd）
Rolling policy: 新しいファイルを作るファイルサイズのしきい値（最小 5 MB）と時間間隔（最小 10 秒）
Credentials: 認証情報の自動生成か、R2 認証情報の手動入力かを選びます
Namespace: Data Catalog の名前空間（デフォルトは default）

セットアップが完了すると、コマンドは Wrangler ファイル用の設定スニペット、サンプルデータ付きの Worker バインディング例、HTTP エンドポイント用の curl コマンドを出力します。以降の手順で使うため、HTTP エンドポイント URL と pipelines 設定を控えておきます。

--name フラグでパイプライン名を事前に指定することもできます。

npx wrangler pipelines setup --name ecommerce

Cloudflare ダッシュボードで R2 object storage を開きます。
Overview を開く ↗
Create bucket を選び、バケット名に pipelines-tutorial を入力します。
Create bucket を選びます。
バケットを選び、Settings タブに切り替えて R2 Data Catalog までスクロールし、Enable を選びます。
有効になったら、Catalog URI と Warehouse name を控えます。
Pipelines > Pipelines を開きます。
Pipelines を開く ↗
Create Pipeline を選びます。
Connect to a Stream:
- Pipeline name: ecommerce
- Enable HTTP endpoint for sending data: 有効
- HTTP authentication: 無効（デフォルト）
- Next を選びます

Define Input Schema:

JSON editor を選びます

次のスキーマをコピーします。

{
	"fields": [
		{
			"name": "user_id",
			"type": "string",
			"required": true
		},
		{
			"name": "event_type",
			"type": "string",
			"required": true
		},
		{
			"name": "product_id",
			"type": "string",
			"required": false
		},
		{
			"name": "amount",
			"type": "float64",
			"required": false
		}
	]
}

Next を選びます

Define Sink:
- R2 バケットを選びます: pipelines-tutorial
- Storage type: R2 Data Catalog
- Namespace: default
- Table name: ecommerce
- Advanced Settings: Maximum Time Interval を 10 seconds に変更します
- Next を選びます
Credentials:
- Automatically create an Account API token for your sink を無効にします
- 手順 1 の Catalog Token を入力します
- Next を選びます
Pipeline Definition:
- デフォルトの SQL クエリはそのままにします。
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream;
```
- Create Pipeline を選びます
パイプライン作成後、次の手順のために Stream ID を控えます。

3. サンプルデータを送信する

パイプラインの HTTP エンドポイントへ e コマースイベントを送信します。

curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -d '[
    {
      "user_id": "user_12345",
      "event_type": "purchase",
      "product_id": "widget-001",
      "amount": 29.99
    },
    {
      "user_id": "user_67890",
      "event_type": "view_product",
      "product_id": "widget-002"
    },
    {
      "user_id": "user_12345",
      "event_type": "add_to_cart",
      "product_id": "widget-003",
      "amount": 15.50
    }
  ]'

{stream-id} は、パイプラインセットアップで得た実際のストリームエンドポイントに置き換えます。

4. バケット内のデータを確認する

Cloudflare ダッシュボードで R2 object storage ページを開きます。
バケット pipelines-tutorial を選びます。
パイプラインが作成した Iceberg のメタデータファイルとデータファイルが表示されます。バケットにファイルが見えない場合は、数分待ってから再試行してください。
データは Apache Iceberg 形式で整理され、メタデータがテーブルのバージョンを追跡します。

5. R2 SQL でデータをクエリする

R2 SQL を使う環境を整えます。

export WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN

または、次の内容で .env ファイルを作成します。

WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_API_TOKEN

YOUR_API_TOKEN は手順 1 で作成したトークンです。環境変数の設定については、Wrangler のシステム環境変数を参照してください。

データをクエリします。

npx wrangler r2 sql query "YOUR_WAREHOUSE_NAME" "
SELECT
    user_id,
    event_type,
    product_id,
    amount
FROM default.ecommerce
WHERE event_type = 'purchase'
LIMIT 10"

YOUR_WAREHOUSE_NAME は、パイプラインセットアップ時に控えた warehouse 名に置き換えます。Cloudflare ダッシュボードの R2 object storage > 対象バケット > Settings > R2 Data Catalog でも確認できます。

Apache Iceberg に対応する任意のエンジンでも、このテーブルをクエリできます。ほかのエンジンを R2 Data Catalog に接続する方法は、Connect to Iceberg engines を参照してください。