Version: 4.1.x

Cassandra sink connector

note

You can download all the Pulsar connectors on download page.

The Cassandra sink connector pulls messages from Pulsar topics to Cassandra or ScyllaDB clusters.

ScyllaDB Compatibility

The Cassandra sink connector is fully compatible with ScyllaDB. ScyllaDB is a drop-in replacement for Cassandra that maintains full CQL protocol compatibility. You can use the same pulsar-io-cassandra connector with both Apache Cassandra and ScyllaDB without any modifications.

The configuration is identical for both databases. For more information about using Pulsar with ScyllaDB, see Streaming Real-Time Chat Messages into ScyllaDB with Apache Pulsar.

Configuration

The configuration of the Cassandra sink connector has the following properties.

Property

Name	Type	Required	Default	Description
`roots`	String	true	" " (empty string)	A comma-separated list of Cassandra hosts to connect to.
`keyspace`	String	true	" " (empty string)	The key space used for writing pulsar messages. Note: `keyspace` should be created prior to a Cassandra sink.
`keyname`	String	true	" " (empty string)	The key name of the Cassandra column family. The column is used for storing Pulsar message keys. If a Pulsar message doesn't have any key associated, the message value is used as the key.
`columnFamily`	String	true	" " (empty string)	The Cassandra column family name. Note: `columnFamily` should be created prior to a Cassandra sink.
`columnName`	String	true	" " (empty string)	The column name of the Cassandra column family. The column is used for storing Pulsar message values.

Example

Before using the Cassandra sink connector, you need to create a configuration file through one of the following methods.

JSON

{
   "configs": {
      "roots": "localhost:9042",
      "keyspace": "pulsar_test_keyspace",
      "columnFamily": "pulsar_test_table",
      "keyname": "key",
      "columnName": "col"
   }
}

YAML

configs:
    roots: "localhost:9042"
    keyspace: "pulsar_test_keyspace"
    columnFamily: "pulsar_test_table"
    keyname: "key"
    columnName: "col"