# Connector Pipelines

> **📝 Note**: This is a Preview feature.

## Overview

SingleStore Connector Pipelines can use Kafka Connect source connectors to stream data from external systems into SingleStore. SingleStore leverages the Kafka Connect ecosystem instead of building native connectors for every data source, which includes multiple pre-built connectors.

## Key Concepts

The [Kafka Connect](https://kafka.apache.org/41/kafka-connect/) framework is an open-source component of Apache Kafka that provides a scalable and reliable method to stream data between systems. Kafka Connect includes:

* Source connectors: Pull data from external systems into Kafka
* Sink connectors: Push data from Kafka to external systems

SingleStore uses Kafka Connect source connectors to ingest data directly into SingleStore tables without requiring an intermediate Kafka cluster. Connector Pipelines support source connectors only. Sink connectors are not supported. For writing data from SingleStore to Kafka, use the SingleStore Kafka sink connector. Connector Pipelines currently support Amazon Kinesis and Google Cloud Pub/Sub as data sources.

## Architecture

## Key Architectural Features

1. Leaf Node Processing: The extractor processes data on leaf nodes rather than the Master Aggregator which reduces load on the aggregator and improves performance.

2. Static Schema Table: By default, when output.value.field is not specified, records are loaded into a three-column table with `value` (`JSON`), `timestamp` (`BIGINT`), and `topic` (`TEXT`). When you specify an output mode, the inferred table instead matches the payload schema. Refer to [Output Modes](https://docs.singlestore.com/#section-id235800860120506.md) for more information.

3. JSON-Based Offset Management: Uses `information_schema.PIPELINES_SOURCE_OFFSETS` table to track offsets in JSON format that supports complex offset structures required by different connectors.

4. Multi-Task Support: Pipelines can spawn multiple tasks for parallel processing when the data source supports partitioning (for example, Kinesis shards).

## Static Schema Table

The extractor produces structured records containing three groups of fields: the user payload, connector-specific metadata (for example, Kinesis shard ID or Pub/Sub message key), and Kafka Connect framework metadata (`topic`, `timestamp`, `key`, `headers`, `sourcePartition`, `sourceOffset`). The default `FORMAT JSON` mode emits the full record as JSON; use an output mode to project only the payload or to merge the payload with selected metadata fields. Refer to [Syntax](https://docs.singlestore.com/#section-id235405759589231.md) for more information.

## How Connector Pipelines Works

When a Connector pipeline is created, SingleStore performs the following:

* Checks the connector class and configuration parameters
* If using `CREATE INFERRED PIPELINE`, automatically creates an inferred table with the static schema
* Launches the extractor process on leaf nodes
* Ingests data from the external data source and loads data into the table
* Stores offset information in `PIPELINES_SOURCE_OFFSETS` to provide at-least-once delivery. To achieve exactly-once results, define a unique key on the target table and use `IGNORE DUPLICATE KEY ERRORS` or `ON DUPLICATE KEY UPDATE` to handle replayed records.

## Deploy Kafka Connect Connectors



Kinesis and Pub/Sub connectors are prepackaged and updated as part of database upgrades. Custom connectors are not supported; contact [SingleStore Support](http://support.singlestore.com) if you need a connector that is not on the supported list.

The following are the supported source connectors:

| Connector                                                                                                                                           | Full class name                                                | Short name                                       |
| --------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------- | ------------------------------------------------ |
| [Amazon Kinesis](https://docs.singlestore.com/cloud/load-data/data-sources/load-data-from-amazon-kinesis-using-a-connector-pipeline.md)             | `com.singlestore.kafka.connect.kinesis.KinesisSourceConnector` | `KinesisSourceConnector`/`KinesisSource`         |
| [Google Cloud Pub/Sub](https://docs.singlestore.com/cloud/load-data/data-sources/load-data-from-google-cloud-pub-sub-using-a-connector-pipeline.md) | `com.google.pubsub.kafka.source.CloudPubSubSourceConnector`    | `CloudPubSubSourceConnector`/`CloudPubSubSource` |

## Enable Connector Pipelines

Connector Pipelines is an experimental feature that must be explicitly enabled. Run the following command to enable this feature:

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

```

> **📝 Note**: This setting must be configured before creating Connector Pipelines and requires the `SUPER` permission. The setting persists across cluster restarts and changes take effect immediately.

Run the following command to confirm whether this feature is enabled.

```sql
SHOW VARIABLES LIKE 'experimental_features_config'

```

```output

+------------------------------+-------------------------------------------------------+
|        Variable_name         |            Value                                      |
+------------------------------+-------------------------------------------------------+
| experimental_features_config | connector_pipelines_enabled=true                      |
+------------------------------+-------------------------------------------------------+

```

After enabling the Connector Pipelines, start the Kafka Connect source connector manually using the provided configuration.

## Syntax

```sql
CREATE [OR REPLACE] [INFERRED] PIPELINE <pipeline_name>
AS LOAD DATA CONNECTOR '<connector>'
CONFIG '<connector_configuration_json>'
CREDENTIALS '<credentials_json>'
[BATCH_INTERVAL <milliseconds>]
[DISABLE OUT_OF_ORDER OPTIMIZATION]
[DISABLE OFFSETS METADATA GC]
INTO { TABLE <table_name> | PROCEDURE <procedure_name> }
[FORMAT { JSON | AVRO }]
[( <field_mapping>, ... )];
```

The first string after `LOAD DATA CONNECTOR` identifies the connector class. Specify the connector class in any of the following equivalent forms:

* Full class name: `com.singlestore.kafka.connect.kinesis.KinesisSourceConnector`
* Short name: `KinesisSourceConnector` (class name without the package)
* Pruned name: `KinesisSource` (short name without the Connector suffix)

Short and pruned names require the connector JAR to include `ServiceLoader` metadata that is compatible with [KIP-898](https://cwiki.apache.org/confluence/spaces/KAFKA/pages/240881354/KIP-898+Modernize+Connect+plugin+discovery). If the connector does not include this metadata, use the full class name.

Use `INTO TABLE` to load extracted records directly into a table. Use `INTO PROCEDURE` to route each batch through a stored procedure for transformation, enrichment, or fan-out to multiple tables. The procedure receives a `QUERY` variable whose columns match the extractor's output for the selected format and enabled `output.*` fields.

Loading `INTO PROCEDURE` can reduce ingestion throughput compared to `INTO TABLE`. Use `INTO PROCEDURE` only when the transformation cannot be expressed as a `SET` clause or column mapping.

## `CONFIG` Parameter

The `CONFIG` parameter must contain a JSON object. Fields in `CONFIG` are connector-specific except for the following framework-level options:

| Option                    | Type     | Default | Description                                                                                                                                                                 |
| ------------------------- | -------- | ------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `tasks.max`               | `int`    | `4`     | Maximum number of tasks the connector spawns for parallel extraction. Set to match the source's partition count. For Google Cloud Pub/Sub, only`tasks.max = 1`is supported. |
| `output.value`            | `bool`   | `true`  | Include the connector's value structure in the output.                                                                                                                      |
| `output.timestamp`        | `bool`   | `true`  | Include the record timestamp (`ms`since`epoch`).                                                                                                                            |
| `output.topic`            | `bool`   | `true`  | Include the internal topic identifier.                                                                                                                                      |
| `output.key`              | `bool`   | `false` | Include the record key.                                                                                                                                                     |
| `output.headers`          | `bool`   | `false` | Include Kafka Connect headers.                                                                                                                                              |
| `output.source.partition` | `bool`   | `false` | Include the source partition object.                                                                                                                                        |
| `output.source.offset`    | `bool`   | `false` | Include the source offset object.                                                                                                                                           |
| `output.value.field`      | `string` | —       | Enable single-value output mode. Refer to[Output Modes](https://docs.singlestore.com/#section-id235800860120506.md)for more information.                                    |
| `output.value.merge`      | `string` | —       | Enable merge output mode. Refer to[Output Modes](https://docs.singlestore.com/#section-id235800860120506.md)for more information.                                           |
| `output.value.schema`     | `string` | —       | Required in`FORMAT AVRO`. Fully-qualified Avro schema class name for the connector value.                                                                                   |
| `output.key.schema`       | `string` | —       | Required in`FORMAT AVRO`when`output.key`is`true`. Fully-qualified Avro schema class name for the connector key.                                                             |

> **📝 Note**: Place sensitive fields, such as access keys, secrets, tokens, and service account JSON, in the `CREDENTIALS` clause rather than `CONFIG`. Values in `CREDENTIALS` are redacted in `SHOW CREATE PIPELINE`, while values in `CONFIG` are stored and displayed in cleartext.

Refer to [`CREATE PIPELINE`](https://docs.singlestore.com/cloud/reference/sql-reference/pipelines-commands/create-pipeline.md) and [`CREATE INFERRED PIPELINE`](https://docs.singlestore.com/cloud/reference/sql-reference/pipelines-commands/create-inferred-pipeline.md) for more information.

## Output Modes

Each connector produces a Kafka Connect `SourceRecord` with top-level fields (`key`, `value`, `timestamp`, `topic`) and connector-specific data inside value. The shape of value depends on the connector:

* Amazon Kinesis wraps the payload with metadata. `value.data` contains the raw payload bytes; `value.partitionKey`, `value.shardId`, `value.sequenceNumber`, etc. are Kinesis metadata.

  Amazon Kinesis produces a `SourceRecord` with the following structure:
  ```json
  {
      "topic": "kinesis-ingest",
      "timestamp": 1787208784705,
      "key": "key-1",
      "value": {
          "data": "<payload bytes, base64 in JSON mode>",
          "partitionKey": "key-1",
          "sequenceNumber": "49653543268136818...",
          "shardId": "shardId-000000000000",
          "streamName": "my-stream",
          "approximateArrivalTimestamp": 1787208784705
      }
  }
  ```
* Google Cloud Pub/Sub exposes the payload directly. `value` is the payload.

  Google Cloud Pub/Sub produces a `SourceRecord` with the following structure:
  ```json
  {
      "topic": "s2-pubsub",
      "timestamp": 1787212384000,
      "key": null,
      "value": "<payload bytes>"
  }
  ```
* Default output: Without an output mode, the pipeline delivers the full record. The user payload is base64-encoded in JSON mode, or in a `BLOB` column in Avro mode. To extract at query time you would need `FROM_BASE64(JSON_EXTRACT_STRING(value, 'data'))`. Use one of the following output modes instead.

  For example, a Kinesis pipeline in `FORMAT JSON` with no output mode delivers records in the following format:
  ```json
  {
      "value": {
          "data": "eyJpZCI6IDEsICJldmVudCI6ICJvcmRlcl9wbGFjZWQiLCAiYW1vdW50IjogOTkuOTV9",
          "partitionKey": "key-1",
          "shardId": "shardId-000000000000",
          "sequenceNumber": "49653543268136818..."
      },
      "timestamp": 1787208784705,
      "topic": "kinesis-ingest"
  }
  ```
  To read the payload, use `FROM_BASE64` and `JSON_EXTRACT_STRING` together. An output mode performs this decoding automatically.

  * Single-value output mode (`output.value.field`): Extracts the payload and outputs it directly. The inferred table matches the payload schema.

    Amazon Kinesis:
    ```sql
    CREATE INFERRED PIPELINE kinesis_orders
    AS LOAD DATA CONNECTOR 'com.singlestore.kafka.connect.kinesis.KinesisSourceConnector'
    CONFIG '{
      "kafka.topic": "orders-topic",
      "kinesis.stream": "orders-stream",
      "kinesis.region": "us-east-1",
      "tasks.max": "3",
      "output.value.field": "data"
    }'
    CREDENTIALS '{
      "aws.access.key.id": "<ACCESS_KEY>",
      "aws.secret.access.key": "<SECRET_KEY>"
    }'
    FORMAT JSON;
    ```
    Pub/Sub (payload is `value`, so `output.value.field` is `""`):
    ```sql
    CREATE INFERRED PIPELINE pubsub_events
    AS LOAD DATA CONNECTOR 'com.google.pubsub.kafka.source.CloudPubSubSourceConnector'
    CONFIG '{
      "kafka.topic": "s2-events",
      "tasks.max": "1",
      "cps.project": "my-gcp-project",
      "cps.subscription": "events-sub",
      "output.value.field": ""
    }'
    CREDENTIALS '{ "gcp.credentials.json": "<SERVICE_ACCOUNT_JSON>" }'
    FORMAT JSON;
    ```
  * Merge output mode (`output.value.merge`): Decodes the payload and exposes it alongside the metadata fields, so both can be mapped to columns.

    Amazon Kinesis:
    ```sql
    CREATE TABLE kinesis_events (
      ingested_at BIGINT,
      shard TEXT,
      payload JSON
    );

    CREATE PIPELINE kinesis_merge_pipe
    AS LOAD DATA CONNECTOR 'com.singlestore.kafka.connect.kinesis.KinesisSourceConnector'
    CONFIG '{
      "kafka.topic": "orders-topic",
      "kinesis.stream": "orders-stream",
      "kinesis.region": "us-east-1",
      "tasks.max": "3",
      "output.value.merge": "data",
      "output.value": "true",
      "output.timestamp": "true",
      "output.key": "false",
      "output.topic": "false"
    }'
    CREDENTIALS '{ "aws.access.key.id": "<ACCESS_KEY>", "aws.secret.access.key": "<SECRET_KEY>" }'
    INTO TABLE kinesis_events
    FORMAT JSON
    (`ingested_at` <- `timestamp`, `shard` <- `value`::`shardId`, `payload` <- `value`::`data`);
    ```
    Google Cloud Pub/Sub:
    ```sql
    CREATE TABLE pubsub_events_merge (
        ingested_at BIGINT,
        payload JSON
    );

    CREATE PIPELINE pubsub_merge_pipe
    AS LOAD DATA CONNECTOR 'com.google.pubsub.kafka.source.CloudPubSubSourceConnector'
    CONFIG '{
        "kafka.topic": "s2-events",
        "tasks.max": "1",
        "cps.project": "my-gcp-project",
        "cps.subscription": "events-sub",
        "output.value.merge": "",
        "output.value": "true",
        "output.timestamp": "true",
        "output.topic": "false"
    }'
    CREDENTIALS '{ "gcp.credentials.json": "<SERVICE_ACCOUNT_JSON>" }'
    INTO TABLE pubsub_events_merge
    FORMAT JSON
    (`ingested_at` <- `timestamp`, `payload` <- `value`);
    ```
* Avro output mode (`FORMAT AVRO`): Produces an Avro record collection with the same three-group structure as JSON mode: user payload, connector metadata, and framework metadata. Avro mode has the following requirements:

  * The source connector must emit records with a consistent schema.
  * Specify `output.value.schema` in `CONFIG`.
  * If `output.key` is `true`, also specify `output.key.schema` in `CONFIG`.
  * The pipeline fails the batch if the extractor encounters a record with an unexpected schema.

  For `KinesisSourceConnector`, add the following fields to `CONFIG`:
  ```json
  {
      "output.key.schema": "com.singlestore.kafka.connect.kinesis.KinesisKey",
      "output.value.schema": "com.singlestore.kafka.connect.kinesis.KinesisValue"
  }
  ```
  Avro supports byte arrays natively, so binary fields are not base64-encoded, unlike in JSON mode.

Select an output mode:

| Requirement                                    | Mode                     | Configuration                                                                     |
| ---------------------------------------------- | ------------------------ | --------------------------------------------------------------------------------- |
| Load data as typed columns                     | Single-value             | `"output.value.field": "data"`for Kinesis or`"output.value.field": ""`for Pub/Sub |
| Load data with connector or framework metadata | Merge                    | `"output.value.merge": "data"`for Kinesis or`"output.value.merge": ""`for Pub/Sub |
| Load the full connector record                 | Default (no output mode) | No output mode configuration                                                      |

Column mapping syntax. Use `::` to project nested fields:

```sql
(
  `ingested_at`   <- `timestamp`,
  `payload`       <- `value`::`data`,
  `partition_key` <- `value`::`partitionKey`,
  `shard_id`      <- `value`::`shardId`
)
```

For deeper access from a `SET` clause, bind the source field to a variable first (JSON built-ins accept only string literals as paths):

```sql
FORMAT JSON
(@raw_payload <- `value`::`data`)
SET
  order_id = JSON_EXTRACT_STRING(@raw_payload, 'id'),
  amount   = JSON_EXTRACT_DOUBLE(@raw_payload, 'amount');
```

## Connector Configuration Reference

The following tables describe the connector-specific `CONFIG` and `CREDENTIALS` fields for the supported source connectors.

## Amazon Kinesis

## CONFIG Fields

| Field            | Description                                                                                            |
| ---------------- | ------------------------------------------------------------------------------------------------------ |
| `kafka.topic`    | Internal topic identifier (any value; not a real Kafka topic).                                         |
| `kinesis.stream` | Amazon Kinesis Data Stream name.                                                                       |
| `kinesis.region` | AWS region containing the stream (for example,`us-east-1`).                                            |
| `tasks.max`      | Number of parallel extraction tasks. Set to the number of shards in the stream for optimal throughput. |

## CREDENTIALS Fields

| Field                   | Description                                                    |
| ----------------------- | -------------------------------------------------------------- |
| `aws.access.key.id`     | AWS access key ID.                                             |
| `aws.secret.access.key` | AWS secret access key.                                         |
| `aws.session.token`     | (Optional) AWS session token when using temporary credentials. |

> **📝 Note**: Place AWS credentials in the `CREDENTIALS` clause. Values in `CREDENTIALS` are redacted in `SHOW CREATE PIPELINE`.

## Google Cloud Pub/Sub

## CONFIG Fields

| Field              | Description                                                                                                    |
| ------------------ | -------------------------------------------------------------------------------------------------------------- |
| `kafka.topic`      | Internal topic identifier (any value).                                                                         |
| `cps.project`      | Google Cloud project ID that contains the Pub/Sub subscription.                                                |
| `cps.subscription` | Pub/Sub subscription name.                                                                                     |
| `tasks.max`        | Must be`1`. Pub/Sub does not expose partitions, so parallel extraction is not supported at the pipeline level. |

## CREDENTIALS Fields

| Field                  | Description                            |
| ---------------------- | -------------------------------------- |
| `gcp.credentials.json` | Google Cloud service account key JSON. |

## Offset Management

## View Offsets

Connector Pipelines use JSON-based offsets stored in `information_schema.PIPELINES_SOURCE_OFFSETS`, which differs from traditional integer-based offsets used by native Kafka pipelines.

```sql
SELECT * FROM information_schema.PIPELINES_SOURCE_OFFSETS
WHERE PIPELINE_NAME = 'connector-pipeline'
```

## Offset Format Examples

The following is the format of `KEY` and `VALUE` of Amazon Kinesis offsets:

```json
KEY: {"shardId":"shardId-XXXX"}
VALUE: {"sequenceNumber":"XXXX"}

```

Amazon Kinesis:

```sql
SOURCE_PARTITION_KEY: {"shardId":"shardId-000000000000"}
EARLIEST_OFFSET / LATEST_OFFSET: {"sequenceNumber":"49674719724351887279438853487463363258610716483094839298"}
```

The following is the format of `KEY` and `VALUE` of Google Cloud Pub/Sub offsets:

```json
KEY:   NULL
VALUE: {"projects/<gcp-project>/subscriptions/<subscription>":"<sequence>"}
```

Google Cloud Pub/Sub:

For example, a pipeline might have the following offset information:

```sql
DATABASE_NAME: test_db
PIPELINE_NAME: sir_venn
BATCH_ID:     40080
TASK_ID:      0
KEY:          NULL
VALUE:        {"projects/singlestore-private/subscriptions/sir_venn_sub":"29"}
```

## Set Offsets

Use `ALTER PIPELINE … SET OFFSETS` to move a Connector Pipeline to a different starting position, for example, to replay from the earliest available record or to resume from a specific per-partition offset captured earlier.

```sql
ALTER PIPELINE <pipeline_name> SET OFFSETS { EARLIEST | LATEST | '<json_source_partition_offset>' };
```

* `EARLIEST` : Starts from the oldest record available in each source partition.
* `LATEST`: Skips existing records and starts from the tail.
* `'<json_source_partition_offset>'`: Specifies a JSON array of `{ "task_id", "key", "value" }` entries, one per source partition. The key and value shapes match the `KEY` and `VALUE` columns in `information_schema.PIPELINES_SOURCE_OFFSETS`.

Amazon Kinesis:

```sql
ALTER PIPELINE kinesis_pipe SET OFFSETS
'[
  {
    "task_id": 0,
    "key":   {"shardId": "shardId-000000000000"},
    "value": {"sequenceNumber": "49673435375909482348849496489467527403761012818415976450"}
  },
  {
    "task_id": 1,
    "key":   {"shardId": "shardId-000000000001"},
    "value": {"sequenceNumber": "49673435375931783094048027112609063122033660698885619730"}
  }
]';
```

Google Cloud Pub/Sub: Pub/Sub offsets use `NULL` for `KEY`:

```sql
ALTER PIPELINE pubsub_pipe SET OFFSETS
'[
  {
    "task_id": 0,
    "key":   null,
    "value": {"projects/singlestore-private/subscriptions/sir_venn_sub": "29"}
  }
]';
```

To replay all records from the earliest available position:

```sql
ALTER PIPELINE pubsub_pipe SET OFFSETS EARLIEST;
```

> **📝 Note**: - Query `information_schema.PIPELINES_SOURCE_OFFSETS` to capture the current `TASK_ID`, `KEY`, and `VALUE` values before altering the offsets, so you can restore them if needed.
> - After the offset change, the pipeline resumes extraction from the specified position in its next batch, regardless of previously processed offsets.
> - `ALTER PIPELINE … SET OFFSET CURSOR '…'` is a separate CDC-only command for MongoDB® and MySQL. It does not apply to Connector Pipelines.

## Manage Connector Pipelines

## Pipeline Lifecycle Operations

Refer to [The Lifecycle of a Pipeline](https://docs.singlestore.com/cloud/load-data/about-singlestore-pipelines/pipeline-concepts/the-lifecycle-of-a-pipeline.md) for more information.

## Check Pipeline Status

```sql
SELECT 
  PIPELINE_NAME,
  STATE,
  CONFIG_JSON
FROM information_schema.PIPELINES
WHERE PIPELINE_NAME = '<connector-pipeline>'
```

The following are the pipeline states:

* Running: Pipeline is actively ingesting data
* Stopped: Pipeline is stopped
* Error: Pipeline encountered an error

> **📝 Note**: Because the Kafka Connect API does not signal when a source has been fully drained, `START PIPELINE ... FOREGROUND` may continue polling after all currently available records have been processed. For deterministic single-pass ingestion, use the `LIMIT` clause on `START PIPELINE FOREGROUND`, or use `INTO PROCEDURE` with an explicit stop condition.

## Configuration Best Practices

## Task Configuration

Parallel processing with `tasks.max`:

* Configure `tasks.max` based on data source partitioning
* For Amazon Kinesis: Set `tasks.max` equal to the number of shards
* Google Cloud Pub/Sub: Set `tasks.max` to `1`. Pub/Sub does not expose partitions, so parallel extraction is not supported at the pipeline level. To scale Pub/Sub ingest, scale on the Pub/Sub side.
* Monitor `TASK_ID` distribution in `PIPELINES_SOURCE_OFFSETS`

```sql
-- Check task distribution
SELECT 
  TASK_ID,
  COUNT(*) as offset_count
FROM information_schema.PIPELINES_SOURCE_OFFSETS
WHERE PIPELINE_NAME = '<pipeline_name>'
GROUP BY TASK_ID;

```

## Security Best Practices

1. Credential Storage: Always use the `CREDENTIALS` parameter for sensitive information, never include passwords in `CONFIG`

2. Network Security: Ensure secure connections to data sources (use SSL/TLS when available)

3. Access Control: Grant minimum required permissions to pipeline users

4. Audit Logging: Enable logging for pipeline operations and monitor access

## Performance Optimization

1. Configure Extraction Parameters: Use `SET` statements to tune the extraction performance:
   ```sql
   SET GLOBAL pipelines_extractor_max_batch_interval_ms = 1000;
   ```

2. Computed Columns: Create computed columns for frequently accessed JSON fields
   ```sql
   ALTER TABLE <pipeline_table>
   ADD COLUMN customer_id AS (JSON_EXTRACT_STRING(record, 'customer_id')) PERSISTED INT;
   ```

3. Indexes: Add indexes on computed columns for better query performance
   ```sql
   CREATE INDEX idx_customer_id ON <pipeline_table>(customer_id)
   ```

4. Monitor Batch Times: Track batch processing time and adjust configuration if needed

5. Offset Progress: Regularly check `PIPELINES_SOURCE_OFFSETS` to ensure offsets are advancing

6. Batch size and heap tuning. Connector Pipeline batches are bounded by the connector's batch.size (default `100` for Kinesis, maximum is `10,000`). Native Kafka's 1M-record batch does not apply. Large batch.size values can exhaust the connector's JVM heap; tune it with the `pipelines_cdc_java_heap_size` engine variable. Example symptom: `java.lang.OutOfMemoryError: Java heap space`

7. Project frequently queried fields into columns. Prefer field mappings (`col <- value::field`) over `JSON_EXTRACT_*` at query time. Add computed columns and indexes on projected fields for large tables.

## Examples

The following examples demonstrate how to ingest data from Amazon Kinesis and Google Pub/Sub.

## Amazon Kinesis Pipeline

The following example demonstrates how to create a basic Connector Pipeline that automatically creates a table with a static schema and then ingests data from Amazon Kinesis into the table.

```sql
-- Enable the experimental feature
SET GLOBAL experimental_features_config = "connector_pipelines_enabled=true";

-- Create Kinesis pipeline
CREATE INFERRED PIPELINE kinesis_pipelines
AS LOAD DATA CONNECTOR 'KinesisSourceConnector'
CONFIG '{
  "kafka.topic": "orders-topic",
  "kinesis.stream": "orders-stream",
  "kinesis.region": "us-east-1",
  "tasks.max": "3",
  "output.value.field": "data"
}'
CREDENTIALS '{
  "aws.access.key.id": "<ACCESS_KEY>",
  "aws.secret.access.key": "<SECRET_KEY>"
}'
FORMAT JSON;

-- Start the pipeline
START PIPELINE kinesis_pipelines;
```

## Static Schema Table

When an inferred Connector Pipeline is created, SingleStore automatically creates a table. With `output.value.field` set to data and a payload shaped like `{"id": <int>, "event": <string>, "amount": <float>}`, the inferred table has the following columns::

```sql
-- Inferred table for kinesis_pipelines (single-value output mode)
CREATE TABLE `kinesis_pipelines` (
  `id`     BIGINT,
  `event`  LONGTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin,
  `amount` DOUBLE,
  SORT KEY `__UNORDERED` (),
  SHARD KEY ()
);

```

If a merge output mode is used, the inferred table includes both payload columns and the selected metadata columns.

## Querying Data

Query the inferred columns:

```sql
-- Extract a specific field
SELECT id, event, amount
FROM kinesis_pipelines
WHERE event = 'order_placed';


```

```output

+----+---------------+--------+
| id | event         | amount |
+----+---------------+--------+
|  1 | order_placed  |  99.95 |
|  2 | order_placed  |  49.5  |
+----+---------------+--------+
```

## Amazon Kinesis Pipeline with Stored Procedure

The following example demonstrates how to use a stored procedure to process and transform incoming Kinesis data before inserting it into a custom table schema. The stored procedure is specified in the `INTO PROCEDURE` clause.

```sql
-- Enable the experimental feature
SET GLOBAL experimental_features_config = "connector_pipelines_enabled=true";

-- Create the target table for parsed records
CREATE TABLE parsed_kinesis_stream (
    partition_key   VARCHAR(256),
    sequence_number VARCHAR(64),
    shard_id        VARCHAR(64),
    data_base64     LONGTEXT,
    parsed_data     AS FROM_BASE64(data_base64) :> JSON PERSISTED JSON,
    inserted_at     DATETIME DEFAULT CURRENT_TIMESTAMP
);

-- Create the stored procedure to process incoming records
DELIMITER //

CREATE OR REPLACE PROCEDURE parse_kinesis_stream(
    batch QUERY(value JSON, timestamp BIGINT, topic VARCHAR(255)
)
AS
BEGIN
    INSERT INTO parsed_kinesis_stream (
        partition_key,
        sequence_number,
        shard_id,
        data_base64
    )
    SELECT
        JSON_EXTRACT_STRING(value, 'partitionKey'),
        JSON_EXTRACT_STRING(value, 'sequenceNumber'),
        JSON_EXTRACT_STRING(value, 'shardId'),
        JSON_EXTRACT_STRING(value, 'data')
    FROM batch;
END //

DELIMITER ;

-- Create a Kinesis pipeline with INTO PROCEDURE
CREATE PIPELINE kinesis_pipeline_to_proc
AS LOAD DATA CONNECTOR 'KinesisSourceConnector'
CONFIG '{
    "aws.access.key.id": "<aws_access_key>",
    "aws.secret.access.key": "<aws_secret_access_key>",
    "kafka.topic": "kinesis-topic",
    "kinesis.stream": "my-kinesis-stream",
    "kinesis.region": "us-east-1",
    "tasks.max": 4
}'
CREDENTIALS '{}'
BATCH_INTERVAL 2500
INTO PROCEDURE parse_kinesis_stream
FORMAT JSON;

-- Start the pipeline
START PIPELINE kinesis_pipeline_to_proc;
```

## Target Table Schema

The target table stores both the raw base64-encoded data and a computed column that automatically decodes it:

| Column            | Type           | Description                                                         |
| ----------------- | -------------- | ------------------------------------------------------------------- |
| `partition_key`   | `VARCHAR(256)` | Kinesis partition key                                               |
| `sequence_number` | `VARCHAR(64)`  | Kinesis sequence number                                             |
| `shard_id`        | `VARCHAR(64)`  | Source shard identifier                                             |
| `data_base64`     | `LONGTEXT`     | Raw base64-encoded payload                                          |
| `parsed_data`     | `JSON`         | Automatically decoded payload (computed)                            |
| `inserted_at`     | `DATETIME`     | Timestamp when record was inserted (defaults to`CURRENT_TIMESTAMP`) |

## Querying Data

Query the decoded data directly using the computed column:

```sql
 -- Query decoded data
SELECT
 partition_key,
 sequence_number,
 parsed_data
FROM parsed_kinesis_stream;

```

```output

+-----------------+----------------------------------------------------------+----------------------------------------------------------------------------------------+
| partition_key   | sequence_number                                          | parsed_data                                                                            |
+-----------------+----------------------------------------------------------+----------------------------------------------------------------------------------------+
| pending-order-1 | 49673494398310566511114890923921219322120066372600856578 | {"customer_id": "22222", "status": "pending", "order_total": 150.00}                   |
| active-order-2  | 49673494398310566511114890924392700391769771888175218690 | {"customer_id": "33333", "status": "active", "order_total": 450.00}                    |
| debug-test      | 49673494398310566511114890671811828899685246464889454594 | {"customer_id": "12345", "event_type": "purchase"}                                     |
| test-key-2      | 49673494398310566511114890486232044182282706583872864258 | {"customer_id": "67890", "event_type": "login", "timestamp": "2026-04-21T10:00:00Z"}   |
+-----------------+----------------------------------------------------------+----------------------------------------------------------------------------------------+

```

## Working with Base64-Encoded Data

The parsed\_data computed column already decodes the base64 payload. Query the JSON fields directly:

```sql
-- Decode and extract in a single query
SELECT
    JSON_EXTRACT_STRING(parsed_data, 'customer_id') AS customer_id,
    JSON_EXTRACT_STRING(parsed_data, 'status') AS status
FROM parsed_kinesis_stream
WHERE JSON_EXTRACT_STRING(parsed_data, 'status') = 'active';

```

```output

+-------------+--------+
| customer_id | status |
+-------------+--------+
| 33333       | active |
+-------------+--------+
```

## Google Cloud Pub/Sub Pipeline

The following example creates an inferred pipeline that loads JSON messages from a Google Cloud Pub/Sub subscription. The `output.value.field` configuration is set to `""`, so the pipeline uses the entire message payload to infer the target table schema.

The payload contains `id`, `event`, and `email` fields. SingleStore infers the corresponding column names and data types and creates the `pubsub_events` table.

```sql
CREATE TABLE `pubsub_events` (
    `id` BIGINT,
    `event` LONGTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin,
    `email` LONGTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin,
    SORT KEY `__UNORDERED` (),
    SHARD KEY ()
);
```

Create the inferred pipeline using the `CloudPubSubSourceConnector`. The `cps.project` and `cps.subscription properties` specify the Google Cloud project and Pub/Sub subscription. The `CREDENTIALS` clause provides the Google Cloud service account credentials.

```sql
CREATE INFERRED PIPELINE pubsub_events
AS LOAD DATA CONNECTOR 'CloudPubSubSourceConnector'
CONFIG '{
  "kafka.topic": "s2-events",
  "tasks.max": "1",
  "cps.project": "my-gcp-project",
  "cps.subscription": "events-sub",
  "output.value.field": ""
}'
CREDENTIALS '{ "gcp.credentials.json": "<SERVICE_ACCOUNT_JSON>" }'
FORMAT JSON;

START PIPELINE pubsub_events;

SELECT id, event, email FROM pubsub_events;

```

```output

+------+----------------+-------------------------+
| id   | event          | email                   |
+------+----------------+-------------------------+
|    1 | signup         | alice@example.com       |
|    2 | page_view      | bob@example.com         |
|    3 | order_placed   | carol@example.com       |
|    4 | signup         | dave@example.com        |
|    5 | item_added     | alice@example.com       |
+------+----------------+-------------------------+
```

***

Modified at: September 22, 2026

Source: [/cloud/load-data/about-singlestore-pipelines/pipeline-concepts/connector-pipelines/](https://docs.singlestore.com/cloud/load-data/about-singlestore-pipelines/pipeline-concepts/connector-pipelines/)

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