ClickPipes Terraform 参考

所有 ClickPipe 类型都可以通过 ClickHouse Terraform 提供商中的 clickhouse_clickpipe 资源创建和管理。本页介绍提供商的设置，以及每种受支持的 ClickPipe 类型的配置示例。

提供商设置

从提供商 v3.14.0 版本开始，已正式支持 ClickPipes。如果您使用的是更早版本，则需要使用 alpha 版本——请查看提供商更新日志了解详情。

将 ClickHouse 提供商添加到 Terraform 配置中：

terraform {
  required_providers {
    clickhouse = {
      source  = "ClickHouse/clickhouse"
      version = ">= 3.14.0"
    }
  }
}

provider "clickhouse" {
  organization_id = var.organization_id
  token_key       = var.token_key
  token_secret    = var.token_secret
}

请参阅管理 API 密钥，了解如何创建可供提供商使用的 API 密钥。

资源概览

clickhouse_clickpipe 资源具有以下顶层参数：

参数	必需	说明
`name`	是	ClickPipe 的名称。
`service_id`	是	ClickHouse Cloud 服务的 ID。
`source`	是	源配置 (每个 ClickPipe 仅有一个 source 块) 。
`destination`	是	目标端配置。
`scaling`	否	副本数量和规格。默认为 1 个副本。
`field_mappings`	否	源列与目标端列之间的自定义字段映射。
`settings`	否	高级 ClickPipe 设置。
`stopped`	否	设置为 `true` 时，将以停止状态创建 ClickPipe。默认为 `false`。

id 和 state 属性为只读，创建后由 ClickHouse Cloud 填充。

目标端

destination 块适用于所有源类型：

destination {
  database      = "default"           # 目标数据库。默认为 "default"。
  table         = "my_table"          # 目标表名。除 CDC 以外的所有数据源均必填。
  managed_table = true                # 由 ClickPipes 创建并管理该表。默认为 true。

  table_definition {
    engine {
      type      = "MergeTree"         # MergeTree、ReplacingMergeTree、SummingMergeTree 或 Null。
    }
    sorting_key   = ["id", "ts"]      # 可选。
    partition_by  = "toYYYYMM(ts)"    # 可选。
  }

  columns {
    name = "id"
    type = "UInt64"
  }

  columns {
    name = "message"
    type = "String"
  }
}

对于 CDC 源 (Postgres、MySQL、MongoDB、BigQuery) ，目标表会根据源 schema 自动创建——通常只需指定 database。

按 ClickPipe 类型划分的示例

Kafka

支持的 type 取值：kafka、confluent、msk、azureeventhub、redpanda、warpstream。

resource "clickhouse_clickpipe" "kafka_clickpipe" {
  name       = "My Kafka ClickPipe"
  service_id = var.service_id

  scaling {
    replicas               = 2
    replica_cpu_millicores = 250
    replica_memory_gb      = 1.0
  }

  source {
    kafka {
      type           = "confluent"
      format         = "JSONEachRow"
      brokers        = "broker.example.com:9092"
      topics         = "my_topic"
      consumer_group = "clickpipes-consumer-group"
      authentication = "PLAIN"

      credentials {
        username = "my_user"
        password = var.kafka_password
      }

      offset {
        strategy = "from_latest"
      }
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "message"
      type = "String"
    }

    columns {
      name = "timestamp"
      type = "DateTime"
    }
  }
}

Kafka 与 Schema Registry

resource "clickhouse_clickpipe" "kafka_avro_clickpipe" {
  name       = "My Kafka Avro ClickPipe"
  service_id = var.service_id

  source {
    kafka {
      type   = "confluent"
      format = "AvroConfluent"
      brokers = "broker.example.com:9092"
      topics  = "my_avro_topic"

      credentials {
        username = "my_user"
        password = var.kafka_password
      }

      schema_registry {
        url            = "https://schema-registry.example.com"
        authentication = "PLAIN"

        credentials {
          username = "sr_user"
          password = var.schema_registry_password
        }
      }
    }
  }

  destination {
    table         = "my_avro_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "event"
      type = "String"
    }
  }
}

Amazon Kinesis

resource "clickhouse_clickpipe" "kinesis_clickpipe" {
  name       = "My Kinesis ClickPipe"
  service_id = var.service_id

  source {
    kinesis {
      format         = "AvroConfluent"
      stream_name    = "my-stream"
      region         = "us-east-1"
      iterator_type  = "TRIM_HORIZON"
      authentication = "IAM_USER"

      schema_registry {
        type               = "glue"
        glue_region        = "us-east-1"
        glue_registry_name = "my-registry"
      }

      access_key {
        access_key_id = var.aws_access_key_id
        secret_key    = var.aws_secret_key
      }
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "message"
      type = "String"
    }
  }
}

使用已上传 Protobuf schema 的 Kinesis

将 format 设置为 Protobuf，并通过 protobuf_schema 提供 Base64 编码的 .proto 文件或序列化后的 FileDescriptorSet。上传 schema 时，请勿配置 schema_registry。

resource "clickhouse_clickpipe" "kinesis_protobuf_clickpipe" {
  name       = "My Kinesis Protobuf ClickPipe"
  service_id = var.service_id

  source {
    kinesis {
      format          = "Protobuf"
      stream_name     = "my-stream"
      region          = "us-east-1"
      iterator_type   = "TRIM_HORIZON"
      authentication  = "IAM_USER"
      protobuf_schema = filebase64("${path.module}/event.proto")

      access_key {
        access_key_id = var.aws_access_key_id
        secret_key    = var.aws_secret_key
      }
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }
  }
}

使用 IAM 角色的 Kinesis

使用 IAM 角色进行身份验证需要有一个运行在 AWS 上的 ClickHouse 服务。

resource "clickhouse_clickpipe" "kinesis_iam_role_clickpipe" {
  name       = "My Kinesis ClickPipe (IAM Role)"
  service_id = var.service_id

  source {
    kinesis {
      format         = "JSONEachRow"
      stream_name    = "my-stream"
      region         = "us-east-1"
      iterator_type  = "LATEST"
      authentication = "IAM_ROLE"
      iam_role       = "arn:aws:iam::123456789012:role/my-kinesis-role"
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "message"
      type = "String"
    }
  }
}

Amazon S3

resource "clickhouse_clickpipe" "s3_clickpipe" {
  name       = "My S3 ClickPipe"
  service_id = var.service_id

  source {
    object_storage {
      type           = "s3"
      url            = "https://my-bucket.s3.amazonaws.com/data/*.json"
      format         = "JSONEachRow"
      authentication = "IAM_USER"

      access_key {
        access_key_id = var.aws_access_key_id
        secret_key    = var.aws_secret_key
      }
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "message"
      type = "String"
    }
  }
}

通过 SQS 实现 S3 持续摄取

如需使用 SQS 队列 (无序模式) 进行持续摄取，请参见配置无序模式中的设置说明。

resource "clickhouse_clickpipe" "s3_continuous_clickpipe" {
  name       = "My S3 Continuous ClickPipe"
  service_id = var.service_id

  source {
    object_storage {
      type           = "s3"
      url            = "https://my-bucket.s3.amazonaws.com/data/*.json"
      format         = "JSONEachRow"
      is_continuous  = true
      queue_url      = "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"
      authentication = "IAM_USER"

      access_key {
        access_key_id = var.aws_access_key_id
        secret_key    = var.aws_secret_key
      }
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "message"
      type = "String"
    }
  }
}

Google Cloud Storage

service_account_key 必须为 GCP 服务账号 JSON 密钥文件内容经过 base64 编码后的结果。

resource "clickhouse_clickpipe" "gcs_clickpipe" {
  name       = "My GCS ClickPipe"
  service_id = var.service_id

  source {
    object_storage {
      type                = "gcs"
      url                 = "gs://my-bucket/data/*.json"
      format              = "JSONEachRow"
      authentication      = "SERVICE_ACCOUNT"
      service_account_key = var.gcs_service_account_key
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "message"
      type = "String"
    }
  }
}

Azure Blob 存储

resource "clickhouse_clickpipe" "abs_clickpipe" {
  name       = "My Azure Blob ClickPipe"
  service_id = var.service_id

  source {
    object_storage {
      type                 = "azureblobstorage"
      azure_container_name = "my-container"
      path                 = "data/*.json"
      format               = "JSONEachRow"
      authentication       = "CONNECTION_STRING"
      connection_string    = var.azure_connection_string
    }
  }

  destination {
    table         = "my_table"
    managed_table = true

    table_definition {
      engine {
        type = "MergeTree"
      }
    }

    columns {
      name = "id"
      type = "UInt64"
    }

    columns {
      name = "message"
      type = "String"
    }
  }
}

Postgres CDC

resource "clickhouse_clickpipe" "postgres_cdc_clickpipe" {
  name       = "My Postgres CDC ClickPipe"
  service_id = var.service_id

  source {
    postgres {
      host     = "postgres.example.com"
      port     = 5432
      database = "mydb"

      credentials {
        username = "postgres_user"
        password = var.postgres_password
      }

      settings {
        replication_mode = "cdc"

        # 可选设置
        sync_interval_seconds              = 60
        pull_batch_size                    = 100000
        allow_nullable_columns             = true
        initial_load_parallelism           = 4
        snapshot_num_rows_per_partition    = 100000
        snapshot_number_of_parallel_tables = 1
      }

      table_mappings {
        source_schema_name = "public"
        source_table       = "users"
        target_table       = "public_users"
      }

      table_mappings {
        source_schema_name = "public"
        source_table       = "orders"
        target_table       = "public_orders"

        # 可选
        excluded_columns       = ["internal_notes"]
        use_custom_sorting_key = true
        sorting_keys           = ["id", "created_at"]
        table_engine           = "ReplacingMergeTree"
      }
    }
  }

  destination {
    database = "default"
  }
}

使用 IAM 角色的 Postgres

使用 IAM 角色进行身份验证需要一个运行在 AWS 上的 ClickHouse 服务。

resource "clickhouse_clickpipe" "postgres_iam_role_clickpipe" {
  name       = "My Postgres CDC ClickPipe (IAM Role)"
  service_id = var.service_id

  source {
    postgres {
      host           = "mydb.cluster.us-east-1.rds.amazonaws.com"
      port           = 5432
      database       = "mydb"
      type           = "rdspostgres"
      authentication = "iam_role"
      iam_role       = "arn:aws:iam::123456789012:role/my-rds-role"

      credentials {
        username = "postgres_user"
      }

      settings {
        replication_mode = "cdc"
      }

      table_mappings {
        source_schema_name = "public"
        source_table       = "orders"
        target_table       = "public_orders"
      }
    }
  }

  destination {
    database = "default"
  }
}

MySQL CDC

resource "clickhouse_clickpipe" "mysql_cdc_clickpipe" {
  name       = "My MySQL CDC ClickPipe"
  service_id = var.service_id

  source {
    mysql {
      host = "mysql.example.com"
      port = 3306
      type = "mysql"

      credentials {
        username = "mysql_user"
        password = var.mysql_password
      }

      settings {
        replication_mode = "cdc"

        # 可选设置
        sync_interval_seconds              = 30
        pull_batch_size                    = 10000
        allow_nullable_columns             = true
        initial_load_parallelism           = 4
        snapshot_num_rows_per_partition    = 100000
        snapshot_number_of_parallel_tables = 2
      }

      table_mappings {
        source_schema_name = "mydb"
        source_table       = "orders"
        target_table       = "mydb_orders"
      }

      table_mappings {
        source_schema_name = "mydb"
        source_table       = "customers"
        target_table       = "mydb_customers"

        # 可选
        excluded_columns       = ["password_hash"]
        use_custom_sorting_key = true
        sorting_keys           = ["id"]
        table_engine           = "ReplacingMergeTree"
      }
    }
  }

  destination {
    database = "default"
  }
}

MongoDB CDC

resource "clickhouse_clickpipe" "mongodb_cdc_clickpipe" {
  name       = "My MongoDB CDC ClickPipe"
  service_id = var.service_id

  source {
    mongodb {
      uri             = "mongodb+srv://cluster0.example.mongodb.net"
      read_preference = "secondaryPreferred"

      credentials {
        username = "mongo_user"
        password = var.mongodb_password
      }

      settings {
        replication_mode = "cdc"

        # 可选设置
        sync_interval_seconds              = 30
        pull_batch_size                    = 500
        snapshot_num_rows_per_partition    = 100000
        snapshot_number_of_parallel_tables = 2
      }

      table_mappings {
        source_database_name = "mydb"
        source_collection    = "users"
        target_table         = "mydb_users"
      }

      table_mappings {
        source_database_name = "mydb"
        source_collection    = "orders"
        target_table         = "mydb_orders"
        table_engine         = "ReplacingMergeTree"
      }
    }
  }

  destination {
    database = "default"
  }
}

BigQuery

service_account_file 必须为经过 base64 编码的 GCP 服务账号 JSON 密钥文件内容。

resource "clickhouse_clickpipe" "bigquery_snapshot_clickpipe" {
  name       = "My BigQuery ClickPipe"
  service_id = var.service_id

  source {
    bigquery {
      snapshot_staging_path = "gs://my-staging-bucket/staging/"

      credentials {
        service_account_file = var.gcp_service_account_key
      }

      settings {
        replication_mode = "snapshot"

        # 可选设置
        initial_load_parallelism           = 4
        snapshot_num_rows_per_partition    = 100000
        snapshot_number_of_parallel_tables = 2
        allow_nullable_columns             = true
      }

      table_mappings {
        source_dataset_name = "my_dataset"
        source_table        = "my_table"
        target_table        = "my_bigquery_table"
      }

      table_mappings {
        source_dataset_name    = "my_dataset"
        source_table           = "another_table"
        target_table           = "another_bigquery_table"
        table_engine           = "ReplacingMergeTree"
        use_custom_sorting_key = true
        sorting_keys           = ["id"]
        excluded_columns       = ["internal_col"]
      }
    }
  }

  destination {
    database = "default"
  }
}

扩缩容

所有 ClickPipe 类型都支持 scaling 块，可用于配置副本数量及其资源分配：

scaling {
  replicas               = 2     # 默认值：1。最大值：10。
  replica_cpu_millicores = 500   # 介于 125 到 2000 之间。
  replica_memory_gb      = 2.0   # 介于 0.5 到 8.0 之间。
}

导入现有 ClickPipes

可使用 service ID 和 ClickPipe ID 将现有 ClickPipes 导入 Terraform 状态：

terraform import clickhouse_clickpipe.example <service_id>:<clickpipe_id>

流式数据接入 ClickPipes

对象存储接入 ClickPipes

ClickPipes for CDC

程序化访问

网络

提供商设置

资源概览

目标端

按 ClickPipe 类型划分的示例

Kafka

Kafka 与 Schema Registry

Amazon Kinesis

使用已上传 Protobuf schema 的 Kinesis

使用 IAM 角色的 Kinesis

Amazon S3

通过 SQS 实现 S3 持续摄取

Google Cloud Storage

Azure Blob 存储

Postgres CDC

使用 IAM 角色的 Postgres

MySQL CDC

MongoDB CDC

BigQuery

扩缩容

导入现有 ClickPipes

​提供商设置

​资源概览

​目标端

​按 ClickPipe 类型划分的示例

​Kafka

​Kafka 与 Schema Registry

​Amazon Kinesis

​使用已上传 Protobuf schema 的 Kinesis

​使用 IAM 角色的 Kinesis

​Amazon S3

​通过 SQS 实现 S3 持续摄取

​Google Cloud Storage

​Azure Blob 存储

​Postgres CDC

​使用 IAM 角色 的 Postgres

​MySQL CDC

​MongoDB CDC

​BigQuery

​扩缩容

​导入现有 ClickPipes

提供商设置

资源概览

目标端

按 ClickPipe 类型划分的示例

Kafka

Kafka 与 Schema Registry

Amazon Kinesis

使用已上传 Protobuf schema 的 Kinesis

使用 IAM 角色的 Kinesis

Amazon S3

通过 SQS 实现 S3 持续摄取

Google Cloud Storage

Azure Blob 存储

Postgres CDC

使用 IAM 角色的 Postgres

MySQL CDC

MongoDB CDC

BigQuery

扩缩容

导入现有 ClickPipes