← Cloudflare Pipelines / pipelines / reference

Terraform

В этом примере показано, как настроить Pipelines и R2 Data Catalog с помощью Terraform, используя Провайдер Cloudflare ↗ (v5.19.0+).

Эта конфигурация создаёт полноценный конвейер данных: R2 bucket с включённым каталогом данных, ограниченный по правам API токен для приёмника, а также ресурсы потока, приёмника и конвейера, которые принимают данные JSON в Apache Iceberg ↗ в таблице.

Предварительные требования

Terraform CLI ↗ >= 1.0
Аккаунт Cloudflare с включёнными R2 и Pipelines
API токен для вашего аккаунта со следующими правами доступа:
- Pipelines - Edit
- Workers R2 Storage - Edit
- Workers R2 Data Catalog - Edit
- Токены API аккаунта - Edit

Общие сведения об использовании Terraform с Cloudflare см. в документация по Terraform.

Ресурсы Terraform

В этом примере используются следующие ресурсы Cloudflare Terraform:

Ресурс	Описание
`cloudflare_r2_bucket` ↗	Создаёт R2 bucket для хранения данных pipeline
`cloudflare_r2_data_catalog` ↗	Включает R2 Data Catalog для bucket
`cloudflare_pipeline_stream` ↗	Создаёт stream, который получает события через HTTP или Worker bindings
`cloudflare_pipeline_sink` ↗	Создаёт sink, который записывает данные в R2 Data Catalog или R2
`cloudflare_pipeline` ↗	Создаёт pipeline с SQL, который соединяет stream и sink
`cloudflare_account_token` ↗	Создаёт ограниченный API-токен для аутентификации sink

Сквозной пример

С 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. Создание ресурсов 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 вместо R2 Data Catalog

Чтобы записывать необработанные файлы Parquet или JSON в R2 вместо таблиц Iceberg, замените ресурс приёмника на приёмник 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, и два 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 эндпоинта stream. Используйте его, чтобы отправлять данные в pipeline:

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

Очистка

Чтобы удалить все ресурсы, созданные этой конфигурацией:

terraform destroy