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

Static Schema Table

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

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

Querying Data

Extract data from the JSON record column using JSON functions:

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