> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.
> You can stream JSON messages from Pub/Sub into ClickHouse using a Google Dataflow template
# Dataflow Pub/Sub to ClickHouse template
export const Image = ({img, alt, size = "lg"}) => {
const normalizedSize = ["sm", "md", "lg"].includes(size) ? size : "lg";
return ;
};
The Pub/Sub to ClickHouse template is a streaming pipeline that reads JSON-encoded messages from a Pub/Sub subscription and writes them into a ClickHouse table.
Messages that fail to parse or fail to map to the target schema are routed to a dead-letter destination: a ClickHouse table, a Pub/Sub topic, or both.
Pipeline requirements
* The source Pub/Sub subscription must exist.
* Messages published to the subscription must be valid JSON.
* The target ClickHouse table must exist, and its column names must match the field names in the JSON payload.
* The ClickHouse host must be accessible from the Dataflow worker machines.
* At least one dead-letter destination (`clickHouseDeadLetterTable` or `deadLetterTopic`) must be provided. If both are provided, failed messages are routed to both destinations simultaneously.
* When `clickHouseDeadLetterTable` is set, the dead-letter table must already exist in ClickHouse with the schema shown in [Dead-letter handling](#dead-letter-handling).
* When `deadLetterTopic` is set, the Pub/Sub topic must already exist.
Template parameters
| Parameter Name | Parameter Description | Required | Notes |
| --------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `inputSubscription` | The Pub/Sub subscription to read messages from. Example: `projects//subscriptions/`. | ✅ | Messages must be JSON-encoded. |
| `clickHouseUrl` | The ClickHouse endpoint URL. Use `https://` for SSL connections (ClickHouse Cloud) or `http://` for non-SSL connections. Example: `https://:8443` or `http://:8123`. | ✅ | For ClickHouse Cloud, use the HTTPS endpoint on port `8443`. |
| `clickHouseDatabase` | The name of the ClickHouse database where the target table resides. Example: `default`. | ✅ | |
| `clickHouseTable` | The name of the ClickHouse table to write data into. | ✅ | The table must exist before running the pipeline. |
| `clickHouseUsername` | The username to authenticate with ClickHouse. | ✅ | |
| `clickHousePassword` | The password to authenticate with ClickHouse. | ✅ | |
| `clickHouseDeadLetterTable` | The ClickHouse table to write failed messages into. Example: `my_table_dead_letter`. | | At least one of `clickHouseDeadLetterTable` or `deadLetterTopic` must be provided. The table must exist with the dead-letter schema shown in [Dead-letter handling](#dead-letter-handling). |
| `deadLetterTopic` | The Pub/Sub topic to publish failed messages to. Example: `projects//topics/`. | | At least one of `clickHouseDeadLetterTable` or `deadLetterTopic` must be provided. Failed payloads are published to the topic with `errorMessage` and `failedAt` set as message attributes. |
| `windowSeconds` | Duration in seconds for time-based batching windows. | | See [Batching and windowing](#batching-and-windowing) for the interaction with `batchRowCount`. If neither is set, combined mode uses defaults of `30s` and `1000` rows. |
| `batchRowCount` | Number of rows to accumulate before flushing to ClickHouse. | | See [Batching and windowing](#batching-and-windowing) for the interaction with `windowSeconds`. |
| `maxInsertBlockSize` | Maximum number of rows per `INSERT` statement sent to ClickHouse. Defaults to `1,000,000`. | | A `ClickHouseIO` option. |
| `maxRetries` | Maximum number of retry attempts for failed ClickHouse inserts. Defaults to `5`. | | A `ClickHouseIO` option. |
| `insertDeduplicate` | Whether to enable deduplication for `INSERT` queries in replicated ClickHouse tables. Defaults to `true`. | | A `ClickHouseIO` option. |
| `insertQuorum` | For `INSERT` queries in replicated tables, wait for the specified number of replicas to acknowledge the write and linearize the data addition. `0` disables quorum writes. | | A `ClickHouseIO` option. Disabled in default server settings. |
| `insertDistributedSync` | If enabled, `INSERT` queries into distributed tables wait until data is sent to all nodes in the cluster. Defaults to `true`. | | A `ClickHouseIO` option. |
Default values for all `ClickHouseIO` parameters can be found in [`ClickHouseIO` Apache Beam Connector](/docs/integrations/connectors/data-ingestion/etl-tools/apache-beam#clickhouseiowrite-parameters).
Pub/Sub messages must be JSON objects whose top-level field names exactly match the column names of the target ClickHouse table.
To map incoming messages onto the target table, the pipeline performs the following at startup:
1. Fetches the schema of the target ClickHouse table.
2. Builds a Beam `Row` schema from that ClickHouse schema.
3. For each incoming Pub/Sub message, parses the JSON payload and assembles a row by reading the fields named in the ClickHouse schema.
JSON field names must exactly match the ClickHouse column names (matching is case-sensitive). Fields in the message that do not correspond to a ClickHouse column are ignored. If a ClickHouse column has no matching field in the JSON payload, the pipeline attempts to write `NULL` for that column — which only succeeds when the column is declared as [`Nullable`](/docs/reference/data-types/nullable). Messages that fail to parse, whose values cannot be coerced into the column type, or that would write `NULL` to a non-nullable column, are routed to the dead-letter destination.
Type conversion
JSON values are coerced into the corresponding ClickHouse column type:
| ClickHouse Type | Notes |
| ---------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------- |
| [`Float32`](/docs/reference/data-types/float) | Parsed via `Float.valueOf`. |
| [`Float64`](/docs/reference/data-types/float) | Parsed via `Double.valueOf`. |
| [`Date`](/docs/reference/data-types/date) | Parsed as an ISO-8601 date string. |
| [`DateTime`](/docs/reference/data-types/datetime) | Parsed as an ISO-8601 datetime string (e.g. `2026-01-15T12:34:56Z`). |
| [`Array(T)`](/docs/reference/data-types/array) | JSON array; each element is converted to the element type `T`. Empty or missing arrays produce an empty array. |
| Integer types (`Int8`/`Int16`/`Int32`/`Int64`, `UInt8`/`UInt16`/`UInt32`/`UInt64`) | Parsed from the JSON number or its string representation. |
| [`String`](/docs/reference/data-types/string) | Used as-is for textual fields; non-textual JSON nodes are serialized to their JSON string form. |
Batching and windowing
Because the pipeline is streaming, incoming rows are accumulated into windows before being flushed to ClickHouse. The windowing strategy is selected from the parameters you provide:
| `windowSeconds` | `batchRowCount` | Behavior |
| --------------- | --------------- | --------------------------------------------------------------------------------------------------------- |
| set | unset | Time-based fixed windows of `windowSeconds`. |
| unset | set | Global window with a count trigger; fires every `batchRowCount` rows. |
| both set | both set | Global window with a combined trigger; fires on whichever condition is met first (time **or** row count). |
| neither set | neither set | Combined mode with defaults: `30` seconds or `1000` rows, whichever comes first. |
Tuning these values lets you trade latency against insert efficiency. Smaller windows reduce end-to-end latency; larger windows produce fewer, larger `INSERT` batches.
Dead-letter handling
Messages that fail JSON parsing, schema mapping, or type coercion are routed to the configured dead-letter destination(s). At least one of `clickHouseDeadLetterTable` or `deadLetterTopic` must be provided; if both are set, failed messages are sent to both.
ClickHouse dead-letter table
When `clickHouseDeadLetterTable` is set, the dead-letter table must already exist with this fixed schema:
| Column | Type | Description |
| --------------- | ---------- | ------------------------------------------------------ |
| `raw_message` | `String` | The original Pub/Sub message payload as UTF-8 text. |
| `error_message` | `String` | The exception message describing why the row failed. |
| `stack_trace` | `String` | The full Java stack trace captured at failure time. |
| `failed_at` | `DateTime` | The processing-time timestamp at which the row failed. |
A minimal definition for a single-node deployment:
```sql theme={null}
CREATE TABLE my_table_dead_letter (
raw_message String,
error_message String,
stack_trace String,
failed_at DateTime
) ENGINE = MergeTree()
ORDER BY failed_at;
```
Adapt the engine and `ORDER BY` clause for your deployment — use `ReplicatedMergeTree` for replicated tables, add `ON CLUSTER` for distributed setups, and adjust partitioning or TTL as needed.
Pub/Sub dead-letter topic
When `deadLetterTopic` is set, each failed message is republished to the topic with:
* **Payload**: the original message bytes.
* **Attribute** `errorMessage`: the exception message captured at failure time.
* **Attribute** `failedAt`: the processing-time timestamp at which the row failed.
This makes it convenient to replay failed messages once the underlying schema or producer issue has been resolved.
Running the template
The Pub/Sub to ClickHouse template is available from the Google Cloud Console.
Be sure to review this document, and specifically the above sections, to fully understand the template's configuration requirements and prerequisites.
Sign in to your Google Cloud Console and search for Dataflow.
1. Press the `CREATE JOB FROM TEMPLATE` button.
2. Once the template form is open, enter a job name and select the desired region.
3. In the `Dataflow Template` input, type `ClickHouse` or `Pub/Sub`, and select the `Pub/Sub to ClickHouse` template.
4. Once selected, the form expands. Fill in:
* The Pub/Sub input subscription, in the form `projects//subscriptions/`.
* The ClickHouse endpoint URL — for ClickHouse Cloud use `https://:8443`.
* The ClickHouse database, target table, username and password.
* At least one dead-letter destination: a ClickHouse table or a Pub/Sub topic (or both).
5. Optionally customize batching (`windowSeconds`, `batchRowCount`) and `ClickHouseIO` tuning parameters, as detailed in the [Template parameters](#template-parameters) section.
Monitor the job
Navigate to the [Dataflow Jobs tab](https://console.cloud.google.com/dataflow/jobs) in your Google Cloud Console to monitor the status of the job. You'll find the job details, including progress and any errors:
The template also emits the following custom metrics under the `PubSubToClickHouse` namespace, viewable from the Dataflow job page:
| Metric | Type | Description |
| ----------------------- | ------------ | ------------------------------------------------------------------------------ |
| `messages-received` | Counter | Total Pub/Sub messages received by the parsing step. |
| `rows-parsed-ok` | Counter | Messages successfully converted to a row and routed to the main output. |
| `rows-parse-failed` | Counter | Messages that failed parsing or schema mapping and were routed to dead-letter. |
| `message-payload-bytes` | Distribution | Distribution of incoming Pub/Sub message payload sizes, in bytes. |
Troubleshooting
Memory limit (total) exceeded error (code 241)
This error occurs when ClickHouse runs out of memory while processing large batches of data. To resolve this issue:
* Increase the instance resources: Upgrade your ClickHouse server to a larger instance with more memory to handle the data processing load.
* Decrease the batch size: Reduce `batchRowCount` (and/or `maxInsertBlockSize`) in your Dataflow job configuration to send smaller chunks of data to ClickHouse, reducing memory consumption per batch.
All messages are going to the dead-letter destination
The most common causes are:
* The JSON field names do not match the ClickHouse column names exactly (matching is case-sensitive).
* A column type cannot be coerced from the JSON value (for example, a non-ISO-8601 string in a `DateTime` column).
* The target table schema has changed since the pipeline started — the schema is fetched once at startup. Restart the job after applying schema changes.
Inspect the `error_message` and `stack_trace` columns of the ClickHouse dead-letter table (or the `errorMessage` attribute on Pub/Sub dead-letter messages) to identify the root cause.
Pipeline starts but no rows arrive in ClickHouse
* Confirm the subscription is receiving messages — check the `messages-received` metric on the Dataflow job page.
* In time-based mode (`windowSeconds` only), rows are flushed only at window boundaries. Lower `windowSeconds` to verify flushes are occurring.
* Verify network reachability between Dataflow workers and the ClickHouse endpoint (firewall, VPC peering, or private service connect).
Template source code
The template's source code is available in:
* [`GoogleCloudPlatform/DataflowTemplates`](https://github.com/GoogleCloudPlatform/DataflowTemplates/tree/main/v2/googlecloud-to-clickhouse) — the upstream Google Cloud Platform repository.
* [`ClickHouse/DataflowTemplates`](https://github.com/ClickHouse/DataflowTemplates) — ClickHouse's fork.