Sign inTry Free

Important

Helios features are now enabled during weekly update windows and are no longer directly tied to SingleStore engine releases. Refer to the release notes to view the latest features available in your Helios cluster.

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

On this page

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:

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:

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.

  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

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

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:

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 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:

START PIPELINE pubsub_pipeline;

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

Verify Pipeline Status

SHOW PIPELINES;

Returns the pipeline name and state.

The following is the detailed query:

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.

Last modified:

Was this article helpful?

On this page

Was this article helpful?

Verification instructions

Note: You must install cosign to verify the authenticity of the SingleStore file.

Use the following steps to verify the authenticity of singlestoredb-server, singlestoredb-toolbox, singlestoredb-studio, and singlestore-client SingleStore files that have been downloaded.

You may perform the following steps on any computer that can run cosign, such as the main deployment host of the cluster.

  1. (Optional) Run the following command to view the associated signature files.

    curl undefined

  2. Download the signature file from the SingleStore release server.

    • Option 1: Click the Download Signature button next to the SingleStore file.

    • Option 2: Copy and paste the following URL into the address bar of your browser and save the signature file.

    • Option 3: Run the following command to download the signature file.

      curl -O undefined

  3. After the signature file has been downloaded, run the following command to verify the authenticity of the SingleStore file.

    echo -n undefined |
        cosign verify-blob --certificate-oidc-issuer https://oidc.eks.us-east-1.amazonaws.com/id/CCDCDBA1379A5596AB5B2E46DCA385BC \
          --certificate-identity https://kubernetes.io/namespaces/freya-production/serviceaccounts/job-worker \
          --bundle undefined \
          --new-bundle-format -
    Verified OK

Try Out This Notebook to See What’s Possible in SingleStore

Get access to other groundbreaking datasets and engage with our community for expert advice.

Sign UpLog In