Kafka Connect Cassandra Connector. This project includes source & sink connectors.

Experimental phase.

• Usage
• Running Examples
• CassandraSource
  • Supported CQL Types
• CassandraSink
• Configuration
  • Connect Properties
  • Source Connect Properties
  • Sink Connect Properties
• Building From Source

The project can be used in two ways:

1. as a library in another project,

   "kafka-connect-cassandra" is published on maven central by Tuplejump. It can be defined as a dependency in the build file. For example, with SBT,

   libraryDependencies += "com.tuplejump" %% "kafka-connect-cassandra" % "0.0.7"
   

2. using jar with Kafka Connect

   Download the jar and copy it to 'KAFKA_HOME/libs'

The example uses CassandraSource and loads data from demo.event_store. This data is then saved in another table demo.event_store_sink using CassandraSink. To run the example, execute examples/start.sh in the project directory.

This script starts Cassandra, Zookeper, Kafka Server and then initiates a Kafka Connect command using the Source and Sink Configuration specified in examples/config. Once the script execution is completed, the data from demo.event_store will be available in demo.event_store_sink. You can insert a few rows in demo.event_store to see this.

To stop the processes started for running the example, execute examples/stop.sh

Note: On running the example repeatedly, an error is thrown that topic demo already exists but it won't block the execution

It polls Cassandra with specified query. Using this, data can be fetched from Cassandra in two modes:

1. bulk
2. timestamp based

The modes change automatically based on the query. For example,

SELECT * FROM userlog ; //bulk 
                                         
SELECT * FROM userlog WHERE ts > previousTime() ; //timestamp based

SELECT * FROM userlog WHERE ts = currentTime() ; //timestamp based

SELECT * FROM userlog WHERE ts >= previousTime() AND  ts <= currentTime() ; //timestamp based

Here, previousTime() and currentTime() are replaced prior to fetching the data.

All the others (BLOB, INET, UUID, TIMEUUID, LIST, SET, MAP, CUSTOM, UDT, TUPLE, SMALLINT, TINYINT, DATE,TIME) are currently NOT supported.

It stores Kafka SinkRecord in Cassandra tables. Currently, we only support STRUCT type in the SinkRecord. The STRUCT can have multiple fields with primitive fieldtypes. We assume one-to-one mapping between the column names in the Cassandra sink table and the field names.

Say, the SinkRecords has the following STRUCT value

{
    'id': 1,
    'username': 'user1',
    'text': 'This is my first tweet'
}

Then the Cassandra table should have the columns - id, username, text

Note: The library does not create the Cassandra tables - users are expected to create those before starting the sink

Refer examples/config for sample configuration files

The project requires SBT to build from source. Execute the following command in the project directory,

sbt package  
sbt assembly // to generate the jar with all the dependencies

This will build against Scala 2.11.7 by default. You can override this with:

sbt -Dscala.version=2.10.6 assembly

Or to build against multiple Scala versions,

sbt +package