# Load Data from Google Cloud Pub/Sub Using a Connector Pipeline

This guide walks through creating a Connector Pipeline that ingests messages from a Google Cloud Pub/Sub subscription into SingleStore.

## Prerequisites

To complete this guide, your environment must meet the following prerequisites:

* Google Cloud account with permission to create Pub/Sub topics, subscriptions, and service accounts in a GCP project.
* Connector Pipelines enabled (experimental feature).

## Part 1: Enable Connector Pipelines

Enable the feature with the following command:

```sql
SET GLOBAL experimental_features_config = "connector_pipelines_enabled=true";
```

This setting must be configured before creating Connector Pipelines. The setting persists across cluster restarts and changes take effect immediately.

Verify the setting:

```sql
SHOW VARIABLES LIKE 'experimental_features_config';
```

## Part 2: Set Up a Google Cloud Pub/Sub Subscription

## Create a Pub/Sub Topic and Subscription

1. Log into the [Google Cloud Console](https://console.cloud.google.com/).

2. Select the project you want to use, or create a new one.

3. In the navigation menu, select **Pub/Sub > Topics**.

4. Select **Create topic**.

5. Enter a topic ID (for example, `singlestore-events`).

6. Leave **Add a default subscription** enabled so a subscription is created automatically. Alternatively, disable it and create a subscription manually in the next step.

7. Select **Create**.

If you created the subscription manually:

1. Navigate to **Pub/Sub > Subscriptions**.

2. Select **Create subscription**.

3. Enter a subscription ID (for example, `singlestore-events-sub`).

4. Select the topic you created in the previous step.

5. Set **Delivery type to Pull** (the connector uses pull delivery).

6. Set the **Acknowledgement deadline** and **Message retention** duration to match your workload. Longer retention lets you re-process messages if the pipeline is stopped.

7. Select **Create**.

Record the following values as they are required when creating the pipeline:

* GCP project ID (for example, `my-gcp-project`)
* Subscription ID (for example, `singlestore-events-sub`)

## Generate GCP Credentials

The Pub/Sub connector authenticates using a Google Cloud service account key.

## Required IAM Roles

Grant the following predefined role to the service account:

* `roles/pubsub.subscriber`: Read messages from Pub/Sub subscriptions.

For finer-grained control, you can grant the individual permissions instead:

* `pubsub.subscriptions.consume`
* `pubsub.subscriptions.get`

## Create a Service Account

1. In the Google Cloud Console navigation menu, select **IAM & Admin > Service Accounts**.

2. Select **Create service account**.

3. Enter a service account name (for example, `singlestore-pubsub-reader`) and select **Create and continue**.

4. In **Grant this service account access to project**, select the role **Pub/Sub Subscriber**.

5. Select **Continue**, then **Done**.

## Grant the Service Account Access to the Subscription

If your organization uses subscription-level IAM (rather than project-level), grant the service account the Pub/Sub Subscriber role directly on the subscription:

1. Navigate to **Pub/Sub > Subscriptions** and select the subscription.

2. Open the **Permissions** panel.

3. Select **Add principal**.

4. Enter the service account's email address.

5. Assign the role **Pub/Sub Subscriber**.

6. Select **Save**.

## Create a Service Account Key

1. In **IAM & Admin > Service Accounts**, select the service account you created.

2. Open the **Keys** tab.

3. Select **Add key > Create new key**.

4. Select **JSON** as the key type.

5. Select **Create**. The JSON key file is downloaded to your machine.

Save the JSON key file securely. Google Cloud does not let you re-download the key.

You will paste the contents of this JSON file into the pipeline's `CREDENTIALS` clause.

## Part 3: Create a SingleStore Database and Pub/Sub Pipeline

## Create the Database

```sql
CREATE DATABASE pubsub_data;
USE pubsub_data;
```

## Deploy the Kafka Connect Connector

The Google Cloud Pub/Sub source connector (`CloudPubSubSourceConnector`) is prepackaged with SingleStore clusters. It is updated as part of database upgrades.

## Create the Pub/Sub Pipeline

Ensure you have the following information:

* GCP project ID
* Subscription ID
* Service account key JSON

```sql
CREATE INFERRED PIPELINE pubsub_pipeline
AS LOAD DATA KAFKACONNECT 'CloudPubSubSourceConnector'
CONFIG '{
  "kafka.topic": "pubsub-topic",
  "cps.project": "my-gcp-project",
  "cps.subscription": "singlestore-events-sub",
  "tasks.max": "1"
}'
CREDENTIALS '{
  "gcp.credentials.json": "<paste-service-account-key-json>"
}'
FORMAT JSON;
```

**Important configuration notes:**

* `CloudPubSubSourceConnector`: Short name for `com.google.pubsub.kafka.source.CloudPubSubSourceConnector`. You can also use the full class name.
* `kafka.topic`: A logical identifier for the data source. It does not require an actual Kafka topic and is not related to your Pub/Sub topic name.
* `cps.project`: The GCP project ID that contains the Pub/Sub subscription.
* `cps.subscription`: The Pub/Sub subscription ID (not the full resource path).
* `tasks.max`: Must be `1`. Parallel extraction is not supported at the pipeline level. To scale Pub/Sub ingest, scale on the Pub/Sub side.
* Credentials placement: The service account key JSON must go in the `CREDENTIALS` parameter under the key `gcp.credentials.json`. Placing it in `CONFIG` exposes the key in `SHOW CREATE PIPELINE`.
* Format: Default is `FORMAT JSON`. `FORMAT AVRO` is also supported.

## Optional CONFIG parameters

| Field                       | Description                                                            | Default                     |
| --------------------------- | ---------------------------------------------------------------------- | --------------------------- |
| `cps.endpoint`              | Pub/Sub endpoint to use.                                               | `pubsub.googleapis.com:443` |
| `cps.useEmulator`           | Set to`true`when connecting to a local Pub/Sub emulator (for testing). | `false`                     |
| `cps.maxBatchSize`          | Maximum number of messages returned per pull.                          | `100`                       |
| `kafka.key.attribute`       | Pub/Sub message attribute to use as the record key.                    | (none)                      |
| `kafka.timestamp.attribute` | Pub/Sub message attribute to use as the record timestamp.              | (none)                      |

## Static Schema Table

When the inferred pipeline is created, SingleStore automatically generates a table matching the connector's output schema. For the default `FORMAT JSON` mode without an output-mode field, the inferred table has this shape:

```sql
CREATE TABLE `pubsub_pipeline` (
  `value` JSON,
  `timestamp` BIGINT(20) DEFAULT NULL,
  `topic` LONGTEXT COLLATE utf8mb4_bin NOT NULL,
  SORT KEY `__UNORDERED` (),
  SHARD KEY ()
);
```

The table contains three columns:

* `value`: The full connector record (`JSON`).
* `timestamp`: The record timestamp in milliseconds since epoch (`BIGINT`).
* `topic`: The internal topic identifier (`TEXT`).

This static schema lets SingleStore ingest data from various sources without predefined table schemas. To project your payload fields into typed columns, use the `output.value.field config` (single-value output mode) or map fields explicitly. Refer to [Connector Pipelines](https://docs.singlestore.com/cloud/load-data/about-singlestore-pipelines/pipeline-concepts/connector-pipelines.md) for more information on output modes and column mapping.

## Start the Pipeline

## Start in the Foreground

To test the pipeline and load existing messages, run:

```
START PIPELINE pubsub_pipeline FOREGROUND;
```

This command runs synchronously and returns when the current batch of messages has been loaded.

> **📝 Note**: Because the Kafka Connect API does not signal when a source is fully drained, `START PIPELINE ... FOREGROUND` for a Pub/Sub pipeline may continue polling after all currently available messages have been processed. For deterministic single-pass ingestion, use the `LIMIT` clause on `START PIPELINE FOREGROUND`.

## Start in the Background

For continuous streaming, run the following command:

```sql
START PIPELINE pubsub_pipeline;
```

This command runs the pipeline in the background, continuously pulling messages from the subscription.

## Verify Pipeline Status

```sql
SHOW PIPELINES;
```

Returns the pipeline name and state.

The following is the detailed query:

```sql
SELECT
  PIPELINE_NAME,
  STATE,
  CONFIG_JSON
FROM information_schema.PIPELINES
WHERE PIPELINE_NAME = 'pubsub_pipeline';
```

Google Cloud Pub/Sub offsets appear in `PIPELINES_SOURCE_OFFSETS` with `KEY` set to `NULL` and `VALUE` containing the Pub/Sub subscription path mapped to the last acknowledged position.

***

Modified at: September 22, 2026

Source: [/cloud/load-data/data-sources/load-data-from-google-cloud-pub-sub-using-a-connector-pipeline/](https://docs.singlestore.com/cloud/load-data/data-sources/load-data-from-google-cloud-pub-sub-using-a-connector-pipeline/)

(An index of the documentation is available at /llms.txt)