Cassandra

Stream Reactor Cassandra Sink Connector is designed to move data from Apache Kafka to Apache Cassandra . It supports Apache Cssandra 3.0 and later, DataStax Enterprise 4.7 and later, and DataStax Astra databases.

Connector Class

The release artifact is called kafka-connect-casandra-sink.

Features

• Database authentication : User-password, LDAP, Kerberos

• Input formats: The connector supports Avro, Json Schema, Protobuf, Json (schemaless) and primitive data formats. For Schema-Registry based formats, like AVRO, Protobuf, Json-schema, it needs the the Scheam Registry settings.

• At least once semantics : The Connect framework stores record offsets in Kafka and resumes from the last committed offset on restart. This ensures reliable delivery but may occasionally result in duplicate record processing during failures

Example

Connection

The connector requires a connection to the database. To enable it, specify the following configuration entries. See Configuration Reference for details.

Security

SSL connection

When the database cluster has enabled client encryption, configure the SSL Keys and certificates:

User-Password or LDAP authentication

When the database is configured with user-password of LDAP authentication, configure the following:

Kerberos authentication

When the database cluster has Kerberos authentication enabled set the following settings:

KCQL support

KCQL (Kafka Connect Query Language) is a SQL-like syntax that provides a declarative way to configure connector behavior. It serves as the primary interface for defining how data flows from source topics to target Cassandra tables.

Purpose and Functionality

KCQL statements define three core aspects of data pipeline configuration:

Data Source and Target: Specifies which Kafka topic serves as the data source and which Cassandra table receives the data.

Field Mapping: Controls how fields from the source topic map to columns in the target table, including transformations and selective field inclusion.

Operational Parameters: Configures advanced settings such as consistency levels, delete operations, time-to-live (TTL) settings, and timestamp handling.

Key Capabilities

• Flexible Field Mapping: Map source fields to target columns with optional transformations

• Consistency Control: Set read and write consistency levels for Cassandra operations

• Delete Operations: Enable or disable delete functionality based on message content

• TTL Management: Configure automatic data expiration using time-to-live settings

Syntax Template

The basic KCQL statement follows this structure:

INSERT INTO: Specifies the target Cassandra keyspace and table where data will be written.

SELECT: Defines field projection and mapping from the source.

FROM: Identifies the source Kafka topic containing the data to be processed.

PROPERTIES: Optional clause for configuring connector behavior such as consistency levels, TTL settings, and operational parameters.

Examples

Table mapping

Kafka to Database Table Mapping

The connector enables mapping fields from Kafka records to database table columns. A single connector can handle multiple topics, where each topic can map to one or more tables. Kafka messages consist of a Key, Value, and Header.

Generic Mapping Syntax

Use the following syntax to map Kafka message fields to database columns:

• Prefix the field with:

  • _value: For fields from the Kafka message's Value component.

  • _key: For fields from the Kafka message's Key component.

Note: If no prefix is provided, _value is assumed, mapping to the Kafka record's Value component.

The record Key and Value can be mapped to specific columns like this, considering a table with row_key and content columns:

To map database columns with whitespace to KCQL projections, follow this format:

Fan-out

You can map a topic's data to multiple database tables using multiple KCQL statements in the same connect.cassandra.kcql configuration. For example:

User defined types

The connector can map complex types from a Kafka record payload into a user-defined type (UDT) column in the database. The incoming data field names must match the database field names.

Consider the following Kafka record format:

and database definition:

The mapping configuration should be:

Using now() function

You can leverage the now() which returns TIMEUUID function in the mapping. Here is how to use it in KCQL projection:

Write Timestamp

To specify an internal write-time timestamp from the database, choose a numeric field from the Kafka record payload. Use the following mapping syntax:

Examples:

Row-Level TTL

To define the time-to-live (TTL) for a database record, you can optionally map a field from the Kafka record payload. Use the following query:

Examples:

These examples demonstrate specifying TTL units such as 'MINUTES', 'HOURS', or 'DAYS', in addition to the default 'SECONDS'.

Converting date and time for a topic

To configure data and time conversion properties for a topic, use the following SQL command:

Codec Parameter Descriptions:

CQL Queries

Advanced CQL Query Configuration for Kafka Records

When a new Kafka record arrives, you have the option to run a custom CQL query. This feature is designed for advanced use cases; typically, the standard Kafka mapping suffices without needing a query. If you specify a query in the topic-to-table mapping, it will take precedence over the default action.

Important: You must include the bound variables used in the query within the mapping column.

Example

Extra settings

Java Driver settings

Using the connector configuration, prefix with

any specific setting you would want to set for the Cassandra client. Refer to the for more information.

Configuration Reference

Error policies

The connector supports .