Kafka Connect Pipelines

Key Concepts

The 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. Kafka Connect Pipelines support source connectors only. Sink connectors are not supported. For writing data from SingleStore to Kafka, use the SingleStore Kafka sink connector. Kafka Connect Pipelines supports Amazon Kinesis as a data source currently. Refer to Amazon Kinesis for more information.

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: Data is automatically loaded into a predefined table structure with three columns:

   • topic (TEXT): Source identifier

   • id (JSON): Unique record identifier

   • record (JSON): Complete record data

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

CONFIG Parameter

The CONFIG parameter must contain a JSON object with the following:

• Required fields:

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

• Common optional fields:

  • tasks.max: The maximum number of parallel tasks. The default value is 4.

• Connector specific fields: Vary by connector type.

Refer to CREATE PIPELINE and CREATE INFERRED PIPELINE for more information.

Offset Format Examples

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

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

Pipeline Lifecycle Operations

Refer to The Lifecycle of a Pipeline for more information.

Check Pipeline Status

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

The following are the pipeline states:

• Running: Pipeline is actively ingesting data

• Stopped: Pipeline is stopped

• Error: Pipeline encountered an error

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

• Monitor TASK_ID distribution in PIPELINES_SOURCE_OFFSETS

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

   Copy

   SET GLOBAL pipelines_extractor_batch_size = 10000;
   SET GLOBAL pipelines_extractor_max_batch_interval_ms = 1000;

2. Computed Columns: Create computed columns for frequently accessed JSON fields

   SQL

   Copy

   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

   Copy

   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

Example: Amazon Kinesis Pipeline

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

-- Enable the experimental feature
SET GLOBAL experimental_features_config = "kafka_connect_enabled=true";

-- Create Kinesis pipeline
CREATE INFERRED PIPELINE kinesis_pipeline
AS LOAD DATA KAFKACONNECT 'kafka-connector'
CONFIG '{
  "connector.class": "com.singlestore.kafka.connect.kinesis.KinesisSourceConnector",
  "aws.access.key.id": "<aws_access_key>",
  "aws.secret.key.id": "<aws_secret_key>",
  "kafka.topic": "kinesis-topic",
  "kinesis.stream": "my-kinesis-stream",
  "kinesis.region": "us-east-1",
  "tasks.max": 4
}'
CREDENTIALS '{}' -- AWS credentials are in CONFIG for Kinesis connector
FORMAT AVRO;

-- Start the pipeline
START PIPELINE kinesis_pipeline;

CREATE TABLE `<pipeline_name>` (
  `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 ()
)

-- Extract a specific field
SELECT
  topic,
  JSON_EXTRACT_STRING(record, 'fieldName') AS field_value
FROM kinesis_pipeline;

-- Filter based on JSON content
SELECT *
FROM kinesis_pipeline
WHERE JSON_EXTRACT_STRING(record, 'status') = 'active';

-- Extract multiple fields
SELECT
  topic,
  JSON_EXTRACT_STRING(record, 'customer_id') AS customer_id,
  JSON_EXTRACT_STRING(record, 'order_id') AS order_id,
  JSON_EXTRACT_STRING(record, 'timestamp') AS event_time
FROM kinesis_pipeline;

-- Create a computed column for better performance
ALTER TABLE kinesis_pipeline
ADD COLUMN status AS (JSON_EXTRACT_STRING(record, 'status')) PERSISTED STRING;

CREATE INDEX idx_status ON kinesis_pipeline(status);

-- Query using the computed column
SELECT *
FROM kinesis_pipeline
WHERE status = 'active';

Example 2: 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.

-- Enable the experimental feature
SET GLOBAL experimental_features_config = "kafka_connect_enabled=true";

-- Create the target table for parsed records
CREATE TABLE parsed_kinesis_stream (
 partitionKey BIGINT,
 sequence_number VARCHAR(255),
 shardId VARCHAR(255),
 data_base64 VARCHAR(255),
 parsed_data AS FROM_BASE64(data_base64) PERSISTED VARCHAR(255)
);

-- Create the stored procedure to process incoming records
CREATE OR REPLACE PROCEDURE parse_kinesis_stream(
 batch QUERY(topic VARCHAR(255), id JSON, record JSON)
)
AS
BEGIN
 INSERT INTO parsed_kinesis_stream (
   partitionKey,
   sequence_number,
   shardId,
   data_base64
 )
 SELECT
   record::$partitionKey,
   record::$sequenceNumber,
   record::$shardId,
   record::$data
 FROM batch;
END;

-- Create Kinesis pipeline with INTO PROCEDURE
CREATE PIPELINE kinesis_pipeline_to_proc
AS LOAD DATA KAFKACONNECT 'kafka-connector'
CONFIG '{
 "connector.class": "com.singlestore.kafka.connect.kinesis.KinesisSourceConnector",
 "aws.access.key.id": "<aws_access_key>",
 "aws.secret.key.id": "<aws_secret_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 AVRO (
 `topic` <- `topic`,
 `id` <- `id`,
 `record` <- `record`
);

-- Start the pipeline
START PIPELINE kinesis_pipeline_to_proc;

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

-- Extract JSON fields from decoded data
SELECT
 partitionKey,
 JSON_EXTRACT_STRING(parsed_data, 'customer_id') AS customer_id,
 JSON_EXTRACT_STRING(parsed_data, 'event_type') AS event_type
FROM parsed_kinesis_stream;

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