Enrico Olivelli

When <a href="https://techcrunch.com/2021/01/27/datastax-acquires-kesque-as-it-gets-into-data-streaming/">DataStax started investing in streaming with Apache Pulsar</a>™, we knew that one of the first things people would want to do was connect existing enterprise data sources to Apache Cassandra™ using Pulsar.

Apache Pulsar has a powerful framework called Pulsar IO to enable this kind of use case, and at DataStax we already had a best-in-class <a href="https://docs.datastax.com/en/kafka/doc/kafka/kafkaIntro.html">Kafka Connect Sink</a> that enables you to store structured data coming from one or more Kafka topics into DataStax Enterprise, Apache Cassandra, and Astra.

In this post, I'll describe how we built a powerful and versatile Pulsar Sink to connect data streams to Cassandra by abstracting the core logic from our Kafka Sink and expanding it to support Pulsar. This resulted in the recent release of our <a href="https://github.com/datastax/pulsar-sink">Pulsar Sink Connector</a> (read the docs <a href="https://docs.datastax.com/en/pulsar-connector/1.4/index.html">here</a>), which provides users with the same power and flexibility that they get with our Kafka connector.

<h2>Streams of structured data</h2>

Complex data flows can lead to the need to manage many types of data; this is reflected in database tables and schemas. However, you also have to deal with the same level of complexity when you exchange data using streaming technologies.

Apache Pulsar offers built-in mechanisms to deal with Schemas and enables you to consistently manage all of the data structures in your information system, by providing:

<ul>
	<li>An embedded Schema registry, no the need for third-party Schema Registries</li>
	<li>Low-level APIs to handle schema information with Messages</li>
	<li>High-level APIs to handle schema information on Records</li>
	<li>Schema evolution support and automatic validation of compatibility during schema modifications</li>
	<li>Native support for Apache Avro, JSON and Google Protobuf</li>
</ul>

With the Cassandra connector you can leverage Schema information in order to manage the contents of the topics and write them to Cassandra.

You can also map an existing Pulsar schema to the schema of the Cassandra table—even when they don’t match perfectly—by:

<ul>
	<li>selecting only a subset of fields</li>
	<li>selecting fields of inner structures in the case of nested types</li>
	<li>applying simple data transformations</li>
	<li>handling Cassandra TTL and TIMESTAMP properties</li>
</ul>

<h2>From Kafka Connect Sink to Pulsar IO Sink</h2>

The Apache Kafka Connect framework is very similar to Apache Pulsar IO in terms of APIs.

The purpose of a Sink is to write to an external system (in this case Cassandra) the data that is coming from the stream.

In order to leverage all of the existing features of the Kafka Cassandra Sink and keep the two projects on feature parity going forward, we created an abstraction over the two systems: the <a href="https://github.com/datastax/messaging-connectors-commons">Messaging Connectors Commons framework</a>.

 
The Messaging Connectors Commons framework implements:

<ul>
	<li>An abstraction over Kafka Connect and Pulsar IO APIs (Record and Schema APIs)</li>
	<li>Cassandra Client Driver handling (configuration, security, and compatibility with a big matrix of Cassandra, DataStax DSE, and Astra)</li>
	<li>Mapping and Data transformations</li>
	<li>Failure handling</li>
	<li>Multi-topic management</li>
</ul>

We created this framework by extracting all of the common business logic from the <a href="https://github.com/datastax/kafka-sink">Kafka Sink</a>.

After this pure engineering work, the road to building the Pulsar Sink was straightforward: we only had to implement the relevant APIs within the Pulsar Sink framework.

<h2>Differences between Kafka Connect and Pulsar IO</h2>

Even if at first glance the APIs are very similar, the Pulsar IO framework offers a slightly different model, with some cool features and some tradeoffs.

<h3>Single Record Acknowledgement</h3>

In Kafka you are exposed to the partitioned nature of a topic and you also have to manually track the offset you have reached, with one value for each partition.

In Pulsar, you process each record independently and concurrently, then you finish by acknowledging the processing or by calling a negative acknowledgement in order to tell Pulsar that the processing failed locally and the record should be reprocessed. This way the framework can handle dispatching records to the various instances of the Sink and it saves the Sink from having to handle all of the failure-tracking work.

In Pulsar, the Sink receives only one record at a time, but with the single acknowledgement feature, you are free to implement your own batching strategies; in Kafka the batching is done at the framework level so you do not have control over the amount of records to process at one time.

<h3>Structured Data into the Key</h3>

In Pulsar the key of a Record is always a String and it cannot contain structured data. But with the Cassandra sink you can interpret the String value as a JSON-encoded structure and access the contents of each field and of nested structures.

<h3>Schema API is lacking type information</h3>

The Pulsar Schema API framework also doesn’t report type information about generic data structures like Kafka Connect.&nbsp; (<a href="https://github.com/apache/pulsar/pull/9614">This GitHub Pull Request</a> is open to add that.)&nbsp;&nbsp;

The <a href="https://github.com/datastax/messaging-connectors-commons">Messaging Connectors Commons framework</a>, which deals with mapping message data to the Cassandra model, needs detailed type information in order to transform each field to the correct Cassandra data type.

Because in Pulsar we don’t have direct access to this in the Schema, we decided to use Java Reflection and to dynamically build the schema. This approach works very well and it supports schema evolution, as it is expected that the data type does not change for a given field.

One problem arises with null values, because in Java from a null value you cannot know the original data type. In this case, we are initially supposing that the field is of the type String, and we allow it to move from String to another datatype as soon as we see a non-null value.

<h2>Sample usage</h2>

Here’s a simple example about how to map your Pulsar Topic to a Cassandra table:

<div class="code">
<div class="code__heading"><button class="code__copy">Copy</button></div>

<div class="code__body">
<pre>
<code class="lang-yaml">configs:
 topics: mytopic 
 ignoreErrors: None
 ......
 topic:
 mytopic:
 mykeyspace:
 mytable:
 mapping: 'id=key.id,name=value.name,age=value.age'
 consistencyLevel: LOCAL_ONE
 ttl: -1
 ttlTimeUnit: SECONDS
 timestampTimeUnit: MICROSECONDS
 nullToUnset: true
 deletesEnabled: true
 codec:
 locale: en_US
 timeZone: UTC
 timestamp: CQL_TIMESTAMP
 date: ISO_LOCAL_DATE
 time: ISO_LOCAL_TIME
 unit: MILLISECONDS

</code>
</pre>
</div>
</div>

&nbsp;

Here we are reading AVRO or JSON encoded messages from the mytopic topic and we are writing to the mykeyspace.mytable table on Cassandra. (The connection, security, and performance configurations are omitted in order to focus on the mapping features.) 
 
You can map multiple topics to multiple tables, and for each pair you can configure a different mapping, selecting fields from the message and binding each field to a column in Cassandra.

You can also store the full payload as a UDT to Cassandra or extract nested inner data structures.

The Sink automatically handles conversions and efficient writing to Cassandra.

<h2>Compatibility with Cassandra versions</h2>

The Cassandra Sink works well with Apache Cassandra, <a href="https://www.datastax.com/products/datastax-enterprise">Datastax Enterprise (DSE)</a>, and <a href="https://www.datastax.com/products/datastax-astra">Astra</a>.

At Datastax we are running tests on CI against a matrix of many different versions of Cassandra and DSE as well as against the cloud Astra service in order to provide 100% compatibility.

<h2>Wrapping up</h2>

I described how we have been able to refactor the existing Kafka Connect Sink for Apache Cassandra and deliver the same level of features to Pulsar IO users.

<a href="https://docs.datastax.com/en/pulsar-connector/1.4/index.html">Datastax Pulsar Cassandra sink</a> enables you to connect your Pulsar cluster to Cassandra and it offers lots of flexibility in mapping the source data to Cassandra tables.

We are maintaining a full compatibility matrix against many versions of DSE, Apache Cassandra and DataStax Astra.&nbsp;&nbsp;

To learn more about how it works please check out <a href="https://github.com/datastax/pulsar-sink">our repository on GitHub</a>, and feel free to post your questions on the <a href="https://github.com/datastax/pulsar-sink/issues">issue tracker</a>.

Want to try out Apache Pulsar?&nbsp;<a href="https://astra.datastax.com/register">Sign up now</a>&nbsp;for&nbsp;<a href="https://www.datastax.com/products/astra-streaming">Astra Streaming</a>, our fully managed Apache Pulsar service. We’ll give you access to its full capabilities entirely free through beta. See for yourself how easy it is to build modern data applications and let us know what you’d like to see to make your experience even better.&nbsp;

Developing an Enterprise-Level Apache Cassandra Sink Connector for Apache Pulsar

Enrico Olivelli

Share

Share

Streams of structured data

From Kafka Connect Sink to Pulsar IO Sink

Differences between Kafka Connect and Pulsar IO

Single Record Acknowledgement

Structured Data into the Key

Schema API is lacking type information

Sample usage

Compatibility with Cassandra versions

Wrapping up

More Technology

How to Build a Crystal Image Search App with Vector Search

Knowledge Graphs for RAG without a GraphDB

How Winweb Built its AI Assistant with DataStax Astra DB and LangChain

Vercel + Astra DB: Get Data into Your GenAI Apps Fast

One-stop Data API for Production GenAI