Terraform 配置

本示例展示如何使用 Cloudflare provider ↗（v5.19.0+）通过 Terraform 配置 Pipelines 和 R2 Data Catalog。

该配置创建完整的数据 pipeline：启用 data catalog 的 R2 存储桶、用于 sink 的范围限定 API 令牌，以及将 JSON 数据摄取到 Apache Iceberg ↗ 表的 stream、sink 和 pipeline 资源。

前提条件

Terraform CLI ↗ >= 1.0
已启用 R2 和 Pipelines 的 Cloudflare 账户
具有以下权限的范围限定到账户的 API 令牌：
- Pipelines - Edit
- Workers R2 Storage - Edit
- Workers R2 Data Catalog - Edit
- Account API Tokens（账户 API 令牌） - Edit

有关将 Terraform 与 Cloudflare 配合使用的一般信息，请参阅 Terraform 文档。

Terraform 资源

本示例使用以下 Cloudflare Terraform 资源：

资源	描述
`cloudflare_r2_bucket` ↗	创建用于存储 pipeline 数据的 R2 存储桶
`cloudflare_r2_data_catalog` ↗	在存储桶上启用 R2 Data Catalog
`cloudflare_pipeline_stream` ↗	创建通过 HTTP 或 Worker 绑定接收事件的 stream
`cloudflare_pipeline_sink` ↗	创建将数据写入 R2 Data Catalog 或 R2 的 sink
`cloudflare_pipeline` ↗	创建连接 stream 和 sink 的 SQL pipeline
`cloudflare_account_token` ↗	创建用于 sink 身份验证的范围限定 API 令牌

端到端示例

安装 terraform ↗ 后，创建一个目录和以下文件。

1. 定义变量和 provider

创建 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. 创建 pipeline 资源

创建 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 sink 而非 R2 Data Catalog

要将原始 Parquet 或 JSON 文件写入 R2 而非 Iceberg 表，请将 sink 资源替换为 R2 sink。这需要 R2 S3 兼容凭据而非 catalog 令牌。

在 variables.tf 中添加 S3 凭据变量：

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

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

在 main.tf 中替换 sink 资源：

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 sink 时，可以从配置中移除 cloudflare_r2_data_catalog、cloudflare_account_token 以及两个 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 输出 stream 端点 URL。使用它向 pipeline 发送数据：

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

清理

要移除此配置创建的所有资源：

terraform destroy