Terraform

この例では、Cloudflare プロバイダー ↗（v5.19.0 以降）を使い、Terraform で Pipelines と R2 Data Catalog を設定します。

設定では、完全なデータパイプラインを作成します。データカタログを有効にした R2 バケット、シンク用のスコープ付き API トークン、そして JSON データを Apache Iceberg ↗ テーブルへ取り込むストリーム、シンク、パイプラインのリソースです。

前提条件

Terraform CLI ↗ >= 1.0
R2 と Pipelines が有効な Cloudflare アカウント
アカウントにスコープした、次の権限を持つ API トークン:
- Pipelines - Edit
- Workers R2 Storage - Edit
- Workers R2 Data Catalog - Edit
- Account API Tokens - Edit

Cloudflare で Terraform を使う一般的な情報は、Terraform のドキュメントを参照してください。

Terraform リソース

この例では、次の Cloudflare Terraform リソースを使います。

リソース	説明
`cloudflare_r2_bucket` ↗	パイプラインデータを保存する R2 バケットを作成する
`cloudflare_r2_data_catalog` ↗	バケットで R2 Data Catalog を有効にする
`cloudflare_pipeline_stream` ↗	HTTP または Worker バインディング経由でイベントを受け取るストリームを作成する
`cloudflare_pipeline_sink` ↗	R2 Data Catalog または R2 にデータを書き込むシンクを作成する
`cloudflare_pipeline` ↗	ストリームをシンクに接続する SQL 付きパイプラインを作成する
`cloudflare_account_token` ↗	シンク認証用のスコープ付き API トークンを作成する

エンドツーエンドの例

terraform ↗ をインストールしたうえで、ディレクトリと次のファイルを作成します。

1. 変数とプロバイダーを定義する

variables.tf を作成します。

terraform {
  required_providers {
    cloudflare = {
      source  = "cloudflare/cloudflare"
      version = "~> 5.19"
    }
  }
}

provider "cloudflare" {
  api_token = var.cloudflare_api_token
}

variable "cloudflare_api_token" {
  type      = string
  sensitive = true
}

variable "cloudflare_account_id" {
  type = string
}

2. パイプラインリソースを作成する

main.tf を作成します。

# --- R2 bucket and Data Catalog ---

resource "cloudflare_r2_bucket" "pipeline_bucket" {
  account_id = var.cloudflare_account_id
  name       = "my-pipeline-bucket"
}

resource "cloudflare_r2_data_catalog" "pipeline_catalog" {
  account_id  = var.cloudflare_account_id
  bucket_name = cloudflare_r2_bucket.pipeline_bucket.name
}

# --- Scoped API token for the sink ---

data "cloudflare_account_api_token_permission_groups_list" "r2_bucket_item_write" {
  account_id = var.cloudflare_account_id
  name       = "Workers R2 Storage Bucket Item Write"
}

data "cloudflare_account_api_token_permission_groups_list" "r2_data_catalog_write" {
  account_id = var.cloudflare_account_id
  name       = "Workers R2 Data Catalog Write"
}

resource "cloudflare_account_token" "sink_token" {
  name       = "pipeline-sink-token"
  account_id = var.cloudflare_account_id

  policies = [{
    effect = "allow"
    permission_groups = [
      { id = data.cloudflare_account_api_token_permission_groups_list.r2_bucket_item_write.result[0].id },
      { id = data.cloudflare_account_api_token_permission_groups_list.r2_data_catalog_write.result[0].id },
    ]
    resources = jsonencode({
      "com.cloudflare.api.account.${var.cloudflare_account_id}" = "*"
    })
  }]
}

# --- Stream ---

resource "cloudflare_pipeline_stream" "my_stream" {
  account_id = var.cloudflare_account_id
  name       = "my_stream"
  format = {
    type = "json"
  }
  schema = {
    fields = [{
      name     = "value"
      type     = "json"
      required = true
    }]
  }
  http = {
    enabled        = true
    authentication = false
    cors           = {}
  }
  worker_binding = {
    enabled = false
  }
}

# --- Sink (R2 Data Catalog) ---

resource "cloudflare_pipeline_sink" "my_sink" {
  account_id = var.cloudflare_account_id
  name       = "my_sink"
  type       = "r2_data_catalog"
  format = {
    type = "parquet"
  }
  schema = {
    fields = []
  }
  config = {
    account_id = var.cloudflare_account_id
    bucket     = cloudflare_r2_bucket.pipeline_bucket.name
    table_name = cloudflare_r2_data_catalog.pipeline_catalog.name
    token      = cloudflare_account_token.sink_token.value
  }
}

# --- Pipeline ---

resource "cloudflare_pipeline" "my_pipeline" {
  account_id = var.cloudflare_account_id
  name       = "my_pipeline"
  sql        = "INSERT INTO ${cloudflare_pipeline_sink.my_sink.name} SELECT * FROM ${cloudflare_pipeline_stream.my_stream.name}"
}

R2 Data Catalog の代わりに R2 シンクを使う

Iceberg テーブルではなく、生の Parquet または JSON ファイルを R2 に書き込む場合は、シンクリソースを R2 シンクに置き換えます。カタログトークンの代わりに、R2 の S3 互換認証情報が必要です。

S3 認証情報用の変数を variables.tf に追加します。

variable "r2_access_key_id" {
  type      = string
  sensitive = true
}

variable "r2_access_key_secret" {
  type      = string
  sensitive = true
}

main.tf のシンクリソースを置き換えます。

resource "cloudflare_pipeline_sink" "my_sink" {
  account_id = var.cloudflare_account_id
  name       = "my_sink"
  type       = "r2"
  format = {
    type = "json"
  }
  schema = {
    fields = []
  }
  config = {
    account_id = var.cloudflare_account_id
    bucket     = cloudflare_r2_bucket.pipeline_bucket.name
    credentials = {
      access_key_id     = var.r2_access_key_id
      secret_access_key = var.r2_access_key_secret
    }
  }
}

R2 シンクを使う場合は、設定から cloudflare_r2_data_catalog、cloudflare_account_token、および 2 つの cloudflare_account_api_token_permission_groups_list データソースを削除できます。

3. 出力を定義する

outputs.tf を作成します。

output "pipeline_id" {
  value = cloudflare_pipeline.my_pipeline.id
}

output "pipeline_status" {
  value = cloudflare_pipeline.my_pipeline.status
}

output "stream_endpoint" {
  value = cloudflare_pipeline_stream.my_stream.endpoint
}

output "sink_id" {
  value = cloudflare_pipeline_sink.my_sink.id
}

4. デプロイする

環境変数を設定します。

export TF_VAR_cloudflare_api_token="<YOUR_API_TOKEN>"
export TF_VAR_cloudflare_account_id="<YOUR_ACCOUNT_ID>"

変更内容の確認には terraform plan、適用には terraform apply を使えます。

terraform init
terraform plan
terraform apply

apply が完了すると、Terraform はストリームのエンドポイント URL を出力します。パイプラインへデータを送るときに使います。

curl -X POST https://<STREAM_ENDPOINT> \
  -H "Content-Type: application/json" \
  -d '[{"value": {"event": "page_view", "user_id": "user_123"}}]'

クリーンアップ

この設定で作成したリソースをすべて削除するには、次を実行します。

terraform destroy