ClickPipes Terraform reference

All ClickPipes types can be created and managed using the clickhouse_clickpipe resource in the ClickHouse Terraform provider. This page covers provider setup and configuration examples for each supported ClickPipe type.

Provider setup

ClickPipes support is generally available from provider version v3.14.0. If you are using an earlier version, you will need an alpha release — check the provider changelog for details.

Add the ClickHouse provider to your Terraform configuration:

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
}

See Managing API keys for instructions on creating an API key to use with the provider.

Resource overview

The clickhouse_clickpipe resource has the following top-level arguments:

Argument	Required	Description
`name`	Yes	Name for the ClickPipe.
`service_id`	Yes	ID of the ClickHouse Cloud service.
`source`	Yes	Source configuration (one source block per ClickPipe).
`destination`	Yes	Destination configuration.
`scaling`	No	Replica count and size. Defaults to 1 replica.
`field_mappings`	No	Custom field mappings between source and destination columns.
`settings`	No	Advanced ClickPipe settings.
`stopped`	No	Set to `true` to create the ClickPipe in a stopped state. Defaults to `false`.

The id and state attributes are read-only and populated by ClickHouse Cloud after creation.

GCP workload identity (Private Preview)

Follow the GCP workload identity guide to retrieve the ClickPipes-managed service account and grant it access to your source before using the workload identity examples below.

Destination

The destination block is shared across all source types:

destination {
  database      = "default"           # Target database. Defaults to "default".
  table         = "my_table"          # Target table name. Required for all sources except CDC.
  managed_table = true                # Let ClickPipes create and manage the table. Defaults to true.

  table_definition {
    engine {
      type      = "MergeTree"         # MergeTree, ReplacingMergeTree, SummingMergeTree, or Null.
    }
    sorting_key   = ["id", "ts"]      # Optional.
    partition_by  = "toYYYYMM(ts)"    # Optional.
  }

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

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

For CDC sources (Postgres, MySQL, MongoDB, BigQuery), destination tables are created automatically from the source schema — only database is typically needed.

Examples by ClickPipe type

Kafka

Supported type values: kafka, confluent, msk, azureeventhub, redpanda, warpstream, and Private Preview gcmk.

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"
    }
  }
}

Google Managed Kafka with GCP workload identity

Grant the ClickPipes service account roles/managedkafka.client before creating the ClickPipe. Google Managed Kafka requires TLS. Kafka ACLs must allow the service account (User:<service-account-email>) READ and DESCRIBE access to the selected topics and consumer groups.

resource "google_project_iam_member" "gcmk_client" {
  project = var.gcp_project_id
  role    = "roles/managedkafka.client"
  member  = "serviceAccount:${data.clickhouse_clickpipes_service_context.service.gcp_workload_identity.principal}"
}

resource "clickhouse_clickpipe" "gcmk_clickpipe" {
  name       = "My Google Managed Kafka ClickPipe"
  service_id = var.service_id

  source = {
    kafka = {
      type           = "gcmk"
      format         = "JSONEachRow"
      brokers        = "bootstrap.kafka.us-central1.managedkafka.example-project.cloud.goog:9092"
      topics         = "events"
      authentication = "SERVICE_ACCOUNT_WORKLOAD_IDENTITY"
    }
  }

  destination = {
    table         = "events"
    managed_table = true
    table_definition = {
      engine = {
        type = "MergeTree"
      }
    }
    columns = [
      {
        name = "id"
        type = "String"
      }
    ]
  }

  field_mappings = [
    {
      source_field      = "id"
      destination_field = "id"
    }
  ]

  depends_on = [google_project_iam_member.gcmk_client]
}

Kafka with 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"
    }
  }
}

Kinesis with an uploaded Protobuf schema

Set format to Protobuf and use protobuf_schema for a base64-encoded .proto file or serialized FileDescriptorSet. Do not configure schema_registry when uploading a schema.

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"
    }
  }
}

Kinesis with IAM role

IAM role authentication requires a ClickHouse service running on AWS.

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"
    }
  }
}

S3 continuous ingestion with SQS

For continuous ingestion using an SQS queue (unordered mode). See Configuring unordered mode for setup instructions.

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 must be the base64-encoded content of a GCP service account JSON key file.

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"
    }
  }
}

GCS with GCP workload identity

After granting the ClickPipes service account the required GCS permissions, replace the source block above with:

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

For unordered ingestion, also set is_continuous = true and use a fully-qualified queue_url in the form projects/<project-id>/subscriptions/<subscription-id>.

Google Cloud Pub/Sub

The following example uses workload identity. Grant the ClickPipes service account the required Pub/Sub permissions before creating the ClickPipe.

resource "google_project_iam_member" "pubsub_access" {
  project = var.gcp_project_id
  role    = "roles/pubsub.editor"
  member  = "serviceAccount:${data.clickhouse_clickpipes_service_context.service.gcp_workload_identity.principal}"
}

resource "clickhouse_clickpipe" "pubsub_clickpipe" {
  name       = "My Pub/Sub ClickPipe"
  service_id = var.service_id

  source = {
    pubsub = {
      format         = "JSONEachRow"
      project_id     = var.gcp_project_id
      topic          = "events"
      authentication = "SERVICE_ACCOUNT_WORKLOAD_IDENTITY"
      seek_type      = "latest"
    }
  }

  destination = {
    table         = "events"
    managed_table = true
    table_definition = {
      engine = {
        type = "MergeTree"
      }
    }
    columns = [
      {
        name = "id"
        type = "String"
      }
    ]
  }

  field_mappings = [
    {
      source_field      = "id"
      destination_field = "id"
    }
  ]

  depends_on = [google_project_iam_member.pubsub_access]
}

Azure Blob Storage

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"

        # Optional settings
        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"

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

  destination {
    database = "default"
  }
}

Postgres with IAM role

IAM role authentication requires a ClickHouse service running on AWS.

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"

        # Optional settings
        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"

        # Optional
        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"

        # Optional settings
        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 must be the base64-encoded content of a GCP service account JSON key file.

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"

        # Optional settings
        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"
  }
}

BigQuery with GCP workload identity

After granting the ClickPipes service account the required BigQuery and GCS permissions, replace the source block above with:

source = {
  bigquery = {
    authentication        = "SERVICE_ACCOUNT_WORKLOAD_IDENTITY"
    project_id            = var.gcp_project_id
    snapshot_staging_path = "gs://my-staging-bucket/staging/"

    settings = {
      replication_mode = "snapshot"
    }

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

Scaling

All ClickPipe types support the scaling block to configure the number of replicas and their resource allocation:

scaling {
  replicas               = 2     # Default: 1. Maximum: 10.
  replica_cpu_millicores = 500   # Between 125 and 2000.
  replica_memory_gb      = 2.0   # Between 0.5 and 8.0.
}

Importing existing ClickPipes

Existing ClickPipes can be imported into Terraform state using both the service ID and the ClickPipe ID:

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

ClickPipes for Streaming

ClickPipes for Object Storage

ClickPipes for CDC

Programmatic access

Security and access

Networking

Provider setup

Resource overview

GCP workload identity (Private Preview)

Destination

Examples by ClickPipe type

Kafka

Google Managed Kafka with GCP workload identity

Kafka with schema registry

Amazon Kinesis

Kinesis with an uploaded Protobuf schema

Kinesis with IAM role

Amazon S3

S3 continuous ingestion with SQS

Google Cloud Storage

GCS with GCP workload identity

Google Cloud Pub/Sub

Azure Blob Storage

Postgres CDC

Postgres with IAM role

MySQL CDC

MongoDB CDC

BigQuery

BigQuery with GCP workload identity

Scaling

Importing existing ClickPipes

​Provider setup

​Resource overview

​GCP workload identity (Private Preview)

​Destination

​Examples by ClickPipe type

​Kafka

​Google Managed Kafka with GCP workload identity

​Kafka with schema registry

​Amazon Kinesis

​Kinesis with an uploaded Protobuf schema

​Kinesis with IAM role

​Amazon S3

​S3 continuous ingestion with SQS

​Google Cloud Storage

​GCS with GCP workload identity

​Google Cloud Pub/Sub

​Azure Blob Storage

​Postgres CDC

​Postgres with IAM role

​MySQL CDC

​MongoDB CDC

​BigQuery

​BigQuery with GCP workload identity

​Scaling

​Importing existing ClickPipes

Provider setup

Resource overview

GCP workload identity (Private Preview)

Destination

Examples by ClickPipe type

Kafka

Google Managed Kafka with GCP workload identity

Kafka with schema registry

Amazon Kinesis

Kinesis with an uploaded Protobuf schema

Kinesis with IAM role

Amazon S3

S3 continuous ingestion with SQS

Google Cloud Storage

GCS with GCP workload identity

Google Cloud Pub/Sub

Azure Blob Storage

Postgres CDC

Postgres with IAM role

MySQL CDC

MongoDB CDC

BigQuery

BigQuery with GCP workload identity

Scaling

Importing existing ClickPipes