← Cloudflare Pipelines / pipelines / reference

Terraform

Tento příklad ukazuje, jak nakonfigurovat Pipelines a R2 Data Catalog pomocí Terraformu s Poskytovatel Cloudflare ↗ (v5.19.0 a novější).

Tato konfigurace vytvoří kompletní datový pipeline: R2 bucket s povoleným R2 Data Catalog, omezený API token pro sink a prostředky streamu, sinku a pipeline, které přijímají data JSON do Apache Iceberg ↗ tabulka.

Předpoklady

Terraform CLI ↗ >= 1.0
Účet Cloudflare s povolenými službami R2 a Pipelines
API token vázaný na váš účet s těmito oprávněními:
- Pipelines - Edit
- Workers R2 Storage - Edit
- Workers R2 Data Catalog - Edit
- API tokeny účtu - Edit

Obecné informace o používání Terraform s Cloudflare najdete v dokumentaci Terraform.

Prostředky Terraformu

Tento příklad používá následující prostředky Cloudflare Terraform:

Prostředek	Popis
`cloudflare_r2_bucket` ↗	Vytvoří bucket R2 pro ukládání dat pipelinu
`cloudflare_r2_data_catalog` ↗	Povolí R2 Data Catalog na bucketu
`cloudflare_pipeline_stream` ↗	Vytvoří stream, který přijímá události přes HTTP nebo Worker bindings
`cloudflare_pipeline_sink` ↗	Vytvoří sink, který zapisuje data do R2 Data Catalog nebo R2
`cloudflare_pipeline` ↗	Vytvoří pipeline pomocí SQL, který propojí stream se sinkem
`cloudflare_account_token` ↗	Vytvoří API token s omezeným rozsahem pro ověřování sinku

Kompletní příklad

S terraform ↗ nainstalováno, vytvořte adresář a následující soubory.

1. Definujte proměnné a poskytovatele

Vytvořit 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. Vytvořte prostředky pipeline

Vytvořit 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}"
}

Použijte R2 sink místo R2 Data Catalog

Chcete-li místo tabulek Iceberg zapisovat do R2 surové soubory Parquet nebo JSON, nahraďte prostředek typu sink R2 sinkem. Vyžaduje to přihlašovací údaje R2 kompatibilní s S3 namísto tokenu katalogu.

Přidejte proměnné pro přihlašovací údaje S3 do variables.tf:

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

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

Nahraďte prostředek sink v 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
    }
  }
}

Při použití R2 sinku můžete odstranit cloudflare_r2_data_catalog, cloudflare_account_token, a oba cloudflare_account_api_token_permission_groups_list zdroje dat z vaší konfigurace.

3. Definujte výstupy

Vytvořit 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. Nasaďte

Nastavte proměnné prostředí:

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

Poté můžete použít terraform plan k zobrazení změn a terraform apply a použijte je:

terraform init
terraform plan
terraform apply

Po dokončení příkazu apply Terraform vypíše URL adresu endpointu streamu. Použijte ji k odesílání dat do pipeline:

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

Úklid

Chcete-li odstranit všechny prostředky vytvořené touto konfigurací:

terraform destroy