Connector Pipelines

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

  Copy

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

  Copy

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

  Copy

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

    Copy

    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

    Copy

    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

    Copy

    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

    Copy

    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

  Copy

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

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

(
  `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):

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

Amazon Kinesis

Google Cloud Pub/Sub

Offset Format Examples

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

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

Amazon Kinesis:

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:

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

Google Cloud Pub/Sub:

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

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

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

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

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

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

Target Table Schema

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

Querying Data

Query the decoded data directly using the computed column:

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

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

-- 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';

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