If you don not have an Astra account, set one up - it is easy and free.

Here is what you need to do ...

1. Create an Astra account: sign up or sign in at astra.datastax.com.
2. Create a database: follow these instructions or be prepared to use a database you may already have. Make sure you have your database name, database username, database password and keyspace name handy.
3. Obtain database credentials: follow these instructions and download your database credentials:

   curl -L "<SECURE_CONNECT_BUNDLE_DOWNLOAD_URL>" > credentials.zip
   

   You need to copy and paste the above command into Terminal and replace <SECURE_CONNECT_BUNDLE_DOWNLOAD_URL> with your secure connect bundle URL. Again, follow the instructions above to find your unique URL.

To store data into Astra, we need to create a table in a keyspace. Since we intend to pull data from Pulsar topic sensor-data into Astra table temperatures_by_sensor, the table columns must correspond to the fields of messages we expect to publish to the topic. Assume that each message has these three fields:

• string literal to identify a sensor
• timestamp of a measurement
• integer temperature value

First, install the CQL shell for Astra:

wget https://downloads.datastax.com/enterprise/cqlsh-astra.tar.gz

tar xvzf cqlsh-astra.tar.gz

export PATH=$PATH:/root/cqlsh-astra/bin

Second, start the CQL shell and connect to Astra:

cqlsh --secure-connect-bundle=credentials.zip \
      --username=<YOUR_DATABASE_USERNAME>

You need to copy and paste the above command into Terminal and replace <YOUR_DATABASE_USERNAME> with your actual database username. You will also be prompted for your database password.

Third, once connected to Astra, create table temperatures_by_sensor by executing this CQL statement in the CQL shell:

CREATE TABLE <YOUR_KEYSPACE>.temperatures_by_sensor (
  sensor TEXT,
  timestamp TIMESTAMP,
  value INT,
  PRIMARY KEY ((sensor),timestamp)
) WITH CLUSTERING ORDER BY (timestamp DESC);

Again, you have to copy and paste the above statement and replace <YOUR_KEYSPACE> with your existing keyspace name in your Astra database. (You can always create additional keyspaces in Astra; one Astra database can have many keyspaces.)

Finally, exit the CQL shell:

EXIT;

Download the tarball:

wget https://downloads.datastax.com/pulsar/cassandra-enhanced-pulsar-sink-1.4.0.tar.gz

Untar the tarball and copy the .nar file to the Pulsar connectors directory:

tar xvfz cassandra-enhanced-pulsar-sink-1.4.0.tar.gz

mkdir -p /usr/share/pulsar/connectors

cp cassandra-enhanced-pulsar-sink-1.4.0/cassandra-enhanced-pulsar-sink-1.4.0.nar /usr/share/pulsar/connectors/cassandra-enhanced-pulsar-sink-1.4.0.nar

Load the connector:

pulsar-admin sinks reload

To configure a new sink, we use the YAML file:

sensor-data-sink.yml

This file is derived from the sample configuration file that comes with the connector distribution. We made changes to specify the sensor-data topic, the credentials.zip file and how messages in topic sensor-data can be mapped to table temperatures_by_sensor with columns sensor, timestamp and value. Look for these lines in the file:

configs:
  ...
  topics: sensor-data
  contactPoints: 
  loadBalancing.localDc: 
  ...
  cloud.secureConnectBundle: /root/credentials.zip  
  ...
  topic:
    sensor-data:
      <YOUR_KEYSPACE>:
        temperatures_by_sensor:
          mapping: 'sensor=value.sensorId,timestamp=value.timestamp,value=value.value'
  ...

There are three more changes that you have to make. In particular, find and replace <YOUR_DATABASE_USERNAME>, <YOUR_DATABASE_PASSWORD>, and <YOUR_KEYSPACE> with your actual values.

Create a sink instance:

pulsar-admin sinks create \
      --name sensor-data-sink \
      --sink-type cassandra-enhanced \
      --inputs persistent://public/default/sensor-data \
      --sink-config-file /root/sensor-data-sink.yml

Here is more information about the options used in the above command:

Run our custom producer to publish 1000 messages to topic sensor-data:

java -cp sensor-data-producer.jar com.example.pulsar.SensorDataProducer

Check out the producer source code:

sensor-data-producer/src/main/java/com/example/pulsar/SensorDataProducer.java

In particular, study how the message schema and topic are specified.

Verify that all the 1000 messages were stored into table temperatures_by_sensor.

Start the CQL shell and connect to Astra:

cqlsh --secure-connect-bundle=credentials.zip \
      --username=<YOUR_DATABASE_USERNAME>

You need to copy and paste the above command into Terminal and replace <YOUR_DATABASE_USERNAME> with your actual database username. You will also be prompted for your database password.

Do not forget to replace <YOUR_KEYSPACE> with your actual keyspace name in the next two commands.

Count the number of rows in the table:

SELECT COUNT(*) FROM <YOUR_KEYSPACE>.temperatures_by_sensor;

Take a look at how the resulting rows in the table look like:

SELECT * FROM <YOUR_KEYSPACE>.temperatures_by_sensor LIMIT 3;

If you see sample data in the output, you were able to successfully connect Pulsar and Astra!

Exit the CQL shell:

EXIT;