Create a Kinesis Data Stream

1. Log into the AWS Management Console.

2. Navigate to Kinesis.

3. Select Data Streams from the left navigation menu.

4. Select Create data stream.

5. Enter a stream name (e.g., my-kinesis-stream).

6. Select the capacity mode:

   • On-demand: Automatically scales based on throughput

   • Provisioned: Specify the number of shards

7. Select Create data stream.

Note the following information for later use:

• Stream name (e.g., my-kinesis-stream)

• AWS Region (e.g., us-east-1)

• Number of shards (for optimal pipeline performance)

Generate AWS Credentials

To access your Kinesis Data Stream, you need AWS credentials with appropriate permissions.

Create the Database

Create a new database to hold your data:

CREATE DATABASE kinesis_data;
USE kinesis_data;

Deploy the Kafka Connect Connector

SingleStore provides full support for custom Kafka Connect connectors and enables organizations to deploy and configure them with complete control over configuration and management. Organizations also handle ongoing maintenance and updates.

Create the Kinesis Pipeline

Use the following information to create your pipeline:

• Stream name: my-kinesis-stream

• AWS Region: us-east-1

• Access Key ID: <your_access_key_id>

• Secret Access Key: <your_secret_access_key>

• Number of shards: (match your Kinesis stream configuration)

Run the following command by replacing the placeholder values with your own:

CREATE INFERRED PIPELINE kinesis_pipeline
AS LOAD DATA KAFKACONNECT 'kafka-connector'
CONFIG '{
  "connector.class": "com.github.jcustenborder.kafka.connect.kinesis.KinesisSourceConnector",
  "aws.access.key.id": "<your_access_key_id>",
  "aws.secret.key.id": "<your_secret_access_key>",
  "kafka.topic": "kinesis-topic",
  "kinesis.stream": "my-kinesis-stream",
  "kinesis.region": "us-east-1",
  "tasks.max": 4
}'
CREDENTIALS '{}'
FORMAT AVRO;

Important configuration notes:

• connector.class: Fully-qualified Java class name of the Kafka Connect source connector

• tasks.max: Set this equal to the number of shards in your Kinesis stream for optimal performance. The default value is 4.

• kafka.topic: A logical identifier for the data source (does not require an actual Kafka topic)

• kinesis.region: AWS region where your Kinesis stream is located

• Credentials: AWS credentials must be placed in the CONFIG parameter for Kinesis. The CREDENTIALS parameter can remain empty

Static Schema Table

When an inferred Kafka Connect Pipeline is created, SingleStore automatically creates a table with a predefined structure:

CREATE TABLE `kinesis_pipeline` (
  `topic` text CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  `id` JSON COLLATE utf8mb4_bin NOT NULL,
  `record` JSON COLLATE utf8mb4_bin NOT NULL,
  SORT KEY `__UNORDERED` (),
  SHARD KEY ()
)

The table contains three columns:

• topic: Source identifier (TEXT)

• id: Unique record identifier (JSON)

• record: Complete record data (JSON)

This static schema allows SingleStore to ingest data from various sources without requiring predefined table schemas.

Start the Pipeline

You can run the pipeline in the foreground or background.

START PIPELINE kinesis_pipeline FOREGROUND;

START PIPELINE kinesis_pipeline;

Verify Pipeline Status

Check the pipeline status:

SHOW PIPELINES;

+---------------------------+---------+
| Pipelines_in_kinesis_data |  State  |
+---------------------------+---------+
|      kinesis_pipeline     | Running |
+---------------------------+---------+

Run the following command to query detailed pipeline information:

SELECT
  PIPELINE_NAME,
  STATE,
  CONFIG_JSON
FROM information_schema.PIPELINES
WHERE PIPELINE_NAME = 'kinesis_pipeline'