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 (connect bundle): 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.
4. Obtain database credentials (user/password): follow these instructions to generate an application token, with role "Database administrator", to access your database (from CQL and from the sink). Make sure you download the CSV file and store it in a secure location. The database username and password are the "client ID" and "client secret" found in the CSV, respectively.

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. Note: the database username and password are the "client ID" and "client secret" found in the token CSV, respectively.

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'
  ...

(Note that some keys in the data sink configuration have no value, becaues the parameters for the connection are all described within the secure connection bundle credentials.zip.)

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. Note: the database username and password are the "client ID" and "client secret" found in the token CSV, respectively.

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;