Real-time data pipeline Astra objects

Navigate to your Astra portal home and click "Create a Stream".

Name the new streaming tenant “webstore-clicks". Choose any cloud provider and region. Click “Create Tenant”.

You will be redirected to your new tenant’s quickstart. Navigate to the “Namespace and Topics” tab at the top of the screen.

Create a new namespace with the name “production”. We are treating namespaces as logical development environments to illustrate how you could create a continuous delivery flow. You could also have namespaces for “development” and “staging”.

The namespaces view will refresh to display your new "production" namespace. Click the “Add Topic” button associated with the "production" namespace. Name your new topic “all-clicks” and leave it as a “Persistent” topic. Click the “Add Topic” button to finish creating the topic.

Create a second new topic. Click the “Add Topic” button associated with the "production" namespace. Name your second new topic “product-clicks” and leave it as a “Persistent” topic. Click the “Add Topic” button to finish creating the topic.

You should have 2 namespaces. The “production” namespace should contain the "all-clicks" and "product-clicks" topics you created. The "default" namespace is automatically created by Pulsar within each new streaming tenant.

From the Astra portal home, click “Create a Database”.

Name the database “webstore-clicks” and the keyspace “click_data”. Choose any cloud provider and region. Click “Create Database”.

The page will refresh with your new token details. Don’t worry about saving the tokens - we will retrieve these later. You can “Esc” or just return to your Astra portal home, where you will see your new streaming tenant and database.

Copy and paste the following CQL statement into the CQL console and press “Enter”. This will create a table in the database to hold our "all-clicks" web click data (ie: the raw data).

CREATE TABLE IF NOT EXISTS click_data.all_clicks (
  click_timestamp bigint,
  url_host text,
  url_protocol text,
  url_path text,
  url_query text,
  browser_type text,
  operating_system text,
  visitor_id uuid,
  PRIMARY KEY ((operating_system, browser_type, url_host, url_path), click_timestamp)
);

Create a second table in the database. Copy and paste the following CQL statement into the CQL console and press “Enter”. This will create a second table in the database to hold our "product-clicks" web click data (ie: the filtered data).

CREATE TABLE click_data.product_clicks (
    catalog_area_name text,
    product_name text,
    click_timestamp timestamp,
    PRIMARY KEY ((catalog_area_name), product_name, click_timestamp)
) WITH CLUSTERING ORDER BY (product_name ASC, click_timestamp DESC);

You can confirm everything was created correctly by describing the keyspace in the CQL terminal.

describe click_data;

token@cqlsh> describe click_data;

CREATE KEYSPACE click_data WITH replication = {'class': 'NetworkTopologyStrategy', 'us-east-1': '3'}  AND durable_writes = true;

CREATE TABLE click_data.all_clicks (
    operating_system text,
    browser_type text,
    url_host text,
    url_path text,
    click_timestamp bigint,
    url_protocol text,
    url_query text,
    visitor_id uuid,
    PRIMARY KEY ((operating_system, browser_type, url_host, url_path), click_timestamp)
) WITH CLUSTERING ORDER BY (click_timestamp ASC)
    AND additional_write_policy = '99PERCENTILE'
    AND bloom_filter_fp_chance = 0.01
    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
    AND comment = ''
    AND compaction = {'class': 'org.apache.cassandra.db.compaction.UnifiedCompactionStrategy'}
    AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
    AND crc_check_chance = 1.0
    AND default_time_to_live = 0
    AND gc_grace_seconds = 864000
    AND max_index_interval = 2048
    AND memtable_flush_period_in_ms = 0
    AND min_index_interval = 128
    AND read_repair = 'BLOCKING'
    AND speculative_retry = '99PERCENTILE';

CREATE TABLE click_data.product_clicks (
    catalog_area_name text,
    product_name text,
    click_timestamp timestamp,
    PRIMARY KEY (catalog_area_name, product_name, click_timestamp)
) WITH CLUSTERING ORDER BY (product_name ASC, click_timestamp DESC)
    AND additional_write_policy = '99PERCENTILE'
    AND bloom_filter_fp_chance = 0.01
    AND caching = {'keys': 'ALL', 'rows_per_partition': 'NONE'}
    AND comment = ''
    AND compaction = {'class': 'org.apache.cassandra.db.compaction.UnifiedCompactionStrategy'}
    AND compression = {'chunk_length_in_kb': '64', 'class': 'org.apache.cassandra.io.compress.LZ4Compressor'}
    AND crc_check_chance = 1.0
    AND default_time_to_live = 0
    AND gc_grace_seconds = 864000
    AND max_index_interval = 2048
    AND memtable_flush_period_in_ms = 0
    AND min_index_interval = 128
    AND read_repair = 'BLOCKING'
    AND speculative_retry = '99PERCENTILE';

From the Astra portal home, click on the “webstore-clicks” streaming tenant.

Navigate to the “Sinks” tab and click “Create Sink”.

Fill in the details about the new sink as follows:

Click “Create” to create the sink. You will be directed back to the Sinks listing where your new sink is initializing. When your new sink is ready, its status will change to “Running”.

Follow the same flow to create a second sink with the following values:

If everything goes smoothly, you should have 2 sinks in a “Running” state.