Starting with the CLI
DeltaStream provides a Command Line Interface (CLI) that you can use to interact with the platform from the terminal. In this guide, we will walk through the steps to build a streaming application with DeltaStream’s CLI. Please refer to Starting with the Web App if you are interested in using DeltaStream’s web application. Once you finish the steps in this guide, you will have hands-on experience with foundational concepts in DeltaStream and will be able to build similar applications. For this guide, we will be using topics in Apache Kafka, but the steps should be the same if you have your data in other streaming stores such as Amazon Kinesis or Redpanda.
We assume you already have created your account and signed into DeltaStream. Also, we assume you already have created an Organization or have joined an existing Organization.
You will accomplish the following steps in this guide for the CLI:
- Connect to your streaming store (in this case, Apache Kafka) by creating a Store in DeltaStream.
- Create your first Database.
- Create Streams and Changelogs for your Kafka topics.
- Create new Streams, Changelogs, and Materialized Views using DeltaStream’s continuous queries.
- (Optional) Share your streaming data with other users in your Organization.
The first step in the guide is to create a Store in DeltaStream. A Store is a streaming store service such as Apache Kafka or Amazon Kinesis where your streaming data resides. You can create a Store in the DeltaStream CLI using the CREATE STORE statement:
<no-db>/<no-store># CREATE STORE MSK
WITH (
'type' = KAFKA ,
'access_region' = 'AWS us-east-1',
'kafka.sasl.hash_function' = SHA512,
'kafka.sasl.password' = '**********',
'kafka.sasl.username' = 'mskconsumer',
'uris'='b-1-public.cmnuseast1datagenmsk.89yho3.c26.kafka.us-east-1.amazonaws.com:9196'
);
<no-db>/msk# SHOW STORES;
Name | Kind | Access Region | Metadata | Owner | Created at | Updated at
-------+-------+---------------+----------+----------+----------------------+-----------------------
msk | Kafka | AWS us-east-1 | {} | sysadmin | 2023-02-20T11:03:18Z | 2023-02-20T11:03:18Z
<no-db>/msk#
Once you create the Store, it will be shown as the default Store in the prompt in the CLI. Note that before you create your first Store, there is no default Store indicated in the prompt. Using the LIST STORES command lets you see the available Stores that you have created in DeltaStream.
A DeltaStream Store is the metadata for your existing streaming store such as Apache Kafka or Amazon Kinesis. As you can see, in declaring the store, we provide required configurations to connect and use the streaming Store from DeltaStream. You can now inspect the Store and print the content of topics in the Store:
<no-db>/msk# SHOW TOPICS;
Topic name
-----------------------
ds_pageviews
ds_pageviews_pb
ds_shipments
ds_users
<no-db>/msk# PRINT TOPIC ds_pageviews;
{"userid":"User_9"} | {"viewtime":1676891565084,"userid":"User_9","pageid":"Page_28"}
{"userid":"User_7"} | {"viewtime":1676891565264,"userid":"User_7","pageid":"Page_16"}
{"userid":"User_7"} | {"viewtime":1676891565404,"userid":"User_7","pageid":"Page_84"}
{"userid":"User_4"} | {"viewtime":1676891565524,"userid":"User_4","pageid":"Page_17"}
{"userid":"User_3"} | {"viewtime":1676891565664,"userid":"User_3","pageid":"Page_47"}
{"userid":"User_5"} | {"viewtime":1676891565784,"userid":"User_5","pageid":"Page_72"}
Databases are used to organize your streaming data in an intuitive namespace. You can create a Database in the CLI using the CREATE DATABASE statement. Once you create a new Database, DeltaStream creates a default Schema, named
public
, in the database. The following shows the statement to create a new Database with the name TestDB
:<no-db>/msk# CREATE DATABASE TestDB;
testdb.public/msk# SHOW DATABASES;
Name | Default | Owner | Created at | Updated at
---------+---------+----------+----------------------+-----------------------
testdb | ✓ | sysadmin | 2023-02-20T11:18:46Z | 2023-02-20T11:18:46Z
testdb.public/msk# SHOW SCHEMAS;
Name | Default | Owner | Created at | Updated at
---------+---------+----------+----------------------+-----------------------
public | ✓ | sysadmin | 2023-02-20T11:18:46Z | 2023-02-20T11:18:46Z
Note that the CLI prompt shows the current Database and Schema.
We will now create Relations on top of our Kafka topics using DeltaStream’s DDL statements. If you want to treat your streaming data in a Topic as an append only Stream where each event is an independent event in your Stream, you define it as a Stream. In this guide, we declare a Stream on the
ds_pageviews
topic since each pageview event is an independent event:testdb.public/msk# CREATE STREAM pageviews (
viewtime BIGINT,
userid VARCHAR,
pageid VARCHAR
) WITH (
'topic'='ds_pageviews',
'value.format'='JSON'
);
testdb.public/msk# SHOW RELATIONS;
Name | Type | Owner | Created at | Updated at
------------+--------+----------+----------------------+-----------------------
pageviews | Stream | sysadmin | 2023-02-20T11:21:58Z | 2023-02-20T11:21:58Z
We then declare a Changelog for the
ds_users
topic. A Changelog indicates that we want to interpret events in a Topic as UPSERT events; therefore, the events should have a primary key, and each event will be interpreted as an insert or update for the given primary key. Use the CREATE CHANGELOG command to declare the users
Changelog:testdb.public/msk# CREATE CHANGELOG users(
registertime BIGINT,
userid VARCHAR,
regionid VARCHAR,
gender VARCHAR,
interests ARRAY<VARCHAR>,
contactinfo STRUCT<phone VARCHAR, city VARCHAR, "state" VARCHAR, zipcode VARCHAR>,
PRIMARY KEY(userid)
) WITH (
'topic'='ds_users',
'key.format'='json',
'key.type'='STRUCT<userid VARCHAR>',
'value.format'='json'
);
testdb.public/msk# SHOW RELATIONS;
Name | Type | Owner | Created at | Updated at
------------+-----------+----------+----------------------+-----------------------
pageviews | Stream | sysadmin | 2023-02-20T11:21:58Z | 2023-02-20T11:21:58Z
users | Changelog | sysadmin | 2023-02-20T11:29:58Z | 2023-02-20T11:29:58Z
Once you have declared Streams and Changelogs, you can write continuous queries in SQL to process this streaming data in real time. Let’s start with a Streaming or Continuous Query where the result of such queries are streamed back to the user. Such queries can be used to inspect your Streams and Changelogs or build queries iteratively by inspecting the query’s result. As an example, let’s inspect the
pageviews
stream using the following interactive query:testdb.public/msk# SELECT * FROM pageviews;
^C to exit
Waiting for sandbox to be in running state 'defined'................
Waiting for interactive query('defined') to be in running state...
Interactive query is running
| {"viewtime":1676893147888,"userid":"User_8","pageid":"Page_39"}
| {"viewtime":1676893148008,"userid":"User_4","pageid":"Page_72"}
| {"viewtime":1676893148148,"userid":"User_7","pageid":"Page_41"}
| {"viewtime":1676893148268,"userid":"User_5","pageid":"Page_98"}
| {"viewtime":1676893148243,"userid":"User_7","pageid":"Page_12"}
| {"viewtime":1676893148369,"userid":"User_1","pageid":"Page_10"}
| {"viewtime":1676893148423,"userid":"User_9","pageid":"Page_12"}
| {"viewtime":1676893148488,"userid":"User_8","pageid":"Page_75"}
While interactive query results are streamed to the user, DeltaStream provides a Streaming or Continuous Query where the query results are stored back in a Store or a Materialized View. As the first step, let’s write a persistent query that joins the pageviews stream with the user’s Changelog to create an enriched pageviews Stream that includes user details for each pageview event. We will also convert the epoch time to a timestamp with a time zone using the
TO_TIMESTAMP_LTZ
function.CREATE STREAM csas_enriched_pv AS
SELECT
TO_TIMESTAMP_LTZ(viewtime, 3) AS viewtime,
p.userid AS userid,
pageid,
TO_TIMESTAMP_LTZ(registertime, 3) AS registertime,
regionid,
gender,
interests,
contactinfo
FROM pageviews p
JOIN users u ON u.userid = p.userid;
Under the hood, DeltaStream will compile and launch the query as an Apache Flink streaming job. You should be able to see the query along with its status using the LIST QUERIES command in the CLI:
testdb.public/msk#
testdb.public/msk# SHOW QUERIES;
ID | State | DSQL | Owner | Created at | Updated at
---------------------------------------+---------+--------------------------------------------------------------+----------+----------------------+-----------------------
a913595a-ad09-452e-81e1-3f440b56fae2 | RUNNING | CREATE STREAM | sysadmin | 2023-02-20T11:42:53Z | 2023-02-20T11:42:53Z
| | csas_enriched_pv AS SELECT | | |
| | TO_TIMESTAMP_LTZ(viewtime, | | |
| | 3) AS viewtime, p.userid | | |
| | AS userid, pageid, | | |
| | TO_TIMESTAMP_LTZ(registertime, | | |
| | 3) AS registertime, regionid, | | |
| | gender, interests, contactinfo | | |
| | FROM pageviews p JOIN users u | | |
| | ON u.userid = p.userid; | | |
testdb.public/msk#
Once the query successfully runs, you will have a new Kafka topic named
csas_enriched_pv
in your Kafka cluster, and a new Stream will also be added to the Streams in your Database, TestDB
. You can examine the content of the new Stream by running the following query from the CLI:testdb.public/msk#
testdb.public/msk# SELECT * FROM csas_enriched_pv;
^C to exit
Interactive query is running
| {"viewtime":"2023-02-20T11:47:05.717Z","userid":"User_6","pageid":"Page_99","registertime":"2023-02-20T11:47:05.676Z","regionid":"Region_6","gender":"FEMALE","interests":["News","Movies"],"contactinfo":{"phone":"9492229999","city":"Irvine","state":"CA","zipcode":"92617"}}
| {"viewtime":"2023-02-20T11:47:05.856Z","userid":"User_5","pageid":"Page_77","registertime":"2023-02-20T11:46:44.951Z","regionid":"Region_4","gender":"OTHER","interests":["Game","Sport"],"contactinfo":{"phone":"6503889999","city":"Palo Alto","state":"CA","zipcode":"94301"}}
| {"viewtime":"2023-02-20T11:47:06.056Z","userid":"User_9","pageid":"Page_24","registertime":"2023-02-20T11:46:47.991Z","regionid":"Region_9","gender":"OTHER","interests":["News","Travel"],"contactinfo":{"phone":"6503349999","city":"San Mateo","state":"CA","zipcode":"94403"}}
| {"viewtime":"2023-02-20T11:47:06.196Z","userid":"User_4","pageid":"Page_45","registertime":"2023-02-20T11:46:36.748Z","regionid":"Region_5","gender":"FEMALE","interests":["Game","Sport"],"contactinfo":{"phone":"4083366881","city":"San Jose","state":"CA","zipcode":"95112"}}
testdb.public/msk#
Now that we have the enriched pageviews stream, let’s build a Materialized View where we compute the number of pageviews per user. Type the following statement in the CLI to create this Materialized View:
CREATE MATERIALIZED VIEW user_view_count AS
SELECT
userid,
COUNT(*) AS view_count
FROM csas_enriched_pv
GROUP BY userid;
Once you run the above query, DeltaStream will launch a streaming job that runs the SELECT statement and materializes the result of the query. You should be able to see this query in the CLI using the LIST QUERIES command:
testdb.public/msk#
testdb.public/msk# SHOW QUERIES;
ID | State | DSQL | Owner | Created at | Updated at
---------------------------------------+---------+--------------------------------------------------------------+----------+----------------------+-----------------------
cf6d2092-ff25-461f-801d-5fbb0d5ceb58 | RUNNING | CREATE MATERIALIZED VIEW | sysadmin | 2023-02-20T12:56:53Z | 2023-02-20T12:56:53Z
| | user_view_count AS SELECT | | |
| | userid, COUNT(*) AS view_count | | |
| | FROM csas_enriched_pv GROUP BY | | |
| | userid; | | |
a913595a-ad09-452e-81e1-3f440b56fae2 | RUNNING | CREATE STREAM | sysadmin | 2023-02-20T11:42:53Z | 2023-02-20T11:42:53Z
| | csas_enriched_pv AS SELECT | | |
| | TO_TIMESTAMP_LTZ(viewtime, | | |
| | 3) AS viewtime, p.userid | | |
| | AS userid, pageid, | | |
| | TO_TIMESTAMP_LTZ(registertime, | | |
| | 3) AS registertime, regionid, | | |
| | gender, interests, contactinfo | | |
| | FROM pageviews p JOIN users u | | |
| | ON u.userid = p.userid; | | |
testdb.public/msk#
The resulting Materialized View then can be queried the same way you would query a Materialized View in traditional relational databases; however, in DeltaStream, the data in the Materialized View is always kept fresh by the streaming job. The following is a simple query to get the current view count for a user with the userid of
User_2
:testdb.public/msk#
testdb.public/msk# SELECT * FROM user_view_count WHERE userid = 'User_2';
userid | view_count
---------+-------------
User_2 | 580
testdb.public/msk# he
As you see, the number of pageviews for
User_2
is 580
at the time of running the above query. Now run the query again—you should see an updated result for the pageview count for the user that indicates that every time you run a query on a Materialized View, you will get the most up-to-date result. DeltaStream will make sure the data in the view is continuously updated using the continuous query that declared the Materialized View. You should see something similar to the following by running the same query on the Materialized View a few seconds later:testdb.public/msk#
testdb.public/msk# SELECT * FROM user_view_count WHERE userid = 'User_2';
userid | view_count
---------+-------------
User_2 | 580
testdb.public/msk#
testdb.public/msk#
testdb.public/msk# SELECT * FROM user_view_count WHERE userid = 'User_2';
userid | view_count
---------+-------------
User_2 | 818
testdb.public/msk#
As you see, the result is updated to
818
from the previous value of 580
.All in all in, we have show how DeltaStream makes it easy to build stream processing applications using our CLI tool. We have created a query that joins
pageviews
and users
and creates a new Stream called csas_enriched_pv
. We also have another query that creates a Materialized View named user_view_count
from csas_enriched_pv
.Now that we have seen a basic use case, let’s clean up our environment. To do so, we first terminate the queries and then drop the created Streams, Changelogs, and Materialized Views. To terminate your queries, go to the Queries page. Then you can go to the corresponding Database and Schema to drop the Streams, Changelogs, and Materialized Views. Note that if there is a query that uses a Stream, Changelog, or Materialized View, you should terminate the query before dropping the Relation.