はじめに

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

最初の R2 バケットを作成し、データカタログを有効にする
パイプラインがデータカタログへ認証するために必要な 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. R2 バケットを作成する

まだログインしていない場合は、次を実行します。
```
npx wrangler login
```

R2 バケットを作成します。

npx wrangler r2 bucket create pipelines-tutorial

Cloudflare ダッシュボードで R2 object storage ページを開きます。
Overview を開く ↗
Create bucket を選びます。
バケット名に pipelines-tutorial を入力します。
Create bucket を選びます。

2. R2 Data Catalog を有効にする

R2 バケットでカタログを有効にします。

npx wrangler r2 bucket catalog enable pipelines-tutorial

このコマンドを実行したら、「Warehouse」と「Catalog URI」を控えてください。あとで使います。

Cloudflare ダッシュボードで R2 object storage ページを開きます。
Overview を開く ↗
バケット pipelines-tutorial を選びます。
Settings タブに切り替え、R2 Data Catalog までスクロールし、Enable を選びます。
有効になったら、Catalog URI と Warehouse name を控えます。

3. 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 を控えます。

4. パイプラインを作成する

まず、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 Table
- R2 bucket name: pipelines-tutorial
- Namespace: default
- Table name: ecommerce
- Catalog API token: 手順 3 のトークンを入力します
- Compression: zstd
- Roll file when size reaches (MB): 100
- Roll file when time reaches (seconds): 10（このチュートリアルでは、データを早く確認できるようにします）
SQL transformation: Use simple ingestion query を選び、次を使います。
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream
```

セットアップが完了したら、最終出力に表示される HTTP エンドポイント URL を控えます。

Cloudflare ダッシュボードで 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": "f64",
			"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 を無効にします
- 手順 3 の Catalog Token を入力します
- Next を選びます
Pipeline Definition:
- デフォルトの SQL クエリはそのままにします。
```
INSERT INTO ecommerce_sink SELECT * FROM ecommerce_stream;
```
- Create Pipeline を選びます
パイプライン作成後、次の手順のために Stream ID を控えます。

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

パイプラインの 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} は、パイプラインセットアップで得た実際のストリームエンドポイントに置き換えます。

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

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

7. 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 は手順 3 で作成したトークンです。環境変数の設定については、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 は、手順 2 の warehouse 名に置き換えます。

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