# ksqlDB for Confluent Cloud Quick Start This quick start gets you up and running with ksqlDB in the Confluent Cloud Console. In this quick start guide, you perform the following steps to create a simple streaming application that tracks the latest location of simulated riders. - [Step 1: Create a ksqlDB cluster](#cloud-ksqldb-quick-start-create-cluster) - [Step 2: Create a stream](#cloud-ksqldb-quick-start-create-stream) - [Step 3: Create materialized views](#cloud-ksqldb-quick-start-create-materialized-views) - [Step 4: Run a push query over the stream](#cloud-ksqldb-quick-start-run-push-query) - [Step 5: Populate the stream with events](#cloud-ksqldb-quick-start-populate-stream) - [Step 6: Run a pull query against the materialized view](#cloud-ksqldb-quick-start-run-pull-query) - [Step 7: Clean up](#cloud-ksqldb-quick-start-clean-up) ## Prerequisites - Access to Confluent Cloud. ## Step 1: Create a ksqlDB cluster 1. Log in to Cloud Console. 2. In the navigation menu, click **Environments** and click the **default** tile. 3. If you don’t have an Apache Kafka® cluster yet, click **Add cluster** and follow [these instructions](../get-started/index.md#cloud-create-kafka-cluster). When you have a Kafka cluster available, click the tile of the cluster where you want ksqlDB to run. 4. In the cluster navigation menu, click **ksqlDB**, and in the **ksqlDB** overview page, click **Add cluster**. #### NOTE If your organization doesn’t have any active ksqlDB clusters, the Confluent Cloud Console doesn’t display the **ksqlDB** menu option. To get started, you can create a ksqlDB cluster by using the REST API or Confluent CLI, for example: ```bash confluent ksql cluster create --credential-identity ``` `` is the user account/principal that the ksqlDB cluster operates under and uses to access your Kafka resources. You should create a service account for this purpose. For more information, see [Manage ksqlDB on Confluent Cloud by using the Confluent CLI](../cp-component/ksqldb-ccloud-cli.md#ksql-cloud-cli) and [Quick Start for ksqlDB Cluster API on Confluent Cloud](ksqldb-cluster-api.md#cloud-ksql-cluster-apis). Once you have an active ksqlDB cluster, the **ksqlDB** menu option is accessible for viewing your active ksqlDB clusters. 5. In the **New cluster** page, click the **My account** tile and click **Continue**. For production workloads, you should create or select a service account. 6. In the **Cluster name** text box, enter a name for your cluster or accept the default cluster name, which resembles “ksqlDB_cluster_0”. Accept the default configuration options. 7. Click **Launch cluster** to create your new ksqlDB cluster. The **ksqlDB** overview page opens and shows your cluster with the **Provisioning** status. It can take a few minutes to complete cluster provisioning. ## Step 2: Create a stream When your ksqlDB cluster shows the **Up** status, you can start running queries. In this step, you create a data stream. A stream associates a schema with an underlying Kafka topic. 1. Click the cluster you just created to open the query editor. 2. Copy the following query into the editor window and click **Run query** to create a stream named “riderLocations”. ```sql CREATE STREAM riderLocations (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE) WITH (kafka_topic='quickstart-locations', value_format='json', partitions=1); ``` Your output should resemble: ```json { "@type": "currentStatus", "statementText": "CREATE STREAM RIDERLOCATIONS (PROFILEID STRING, LATITUDE DOUBLE, LONGITUDE DOUBLE) WITH (CLEANUP_POLICY='delete', KAFKA_TOPIC='quickstart-locations', KEY_FORMAT='KAFKA', PARTITIONS=1, VALUE_FORMAT='JSON');", "commandId": "stream/`RIDERLOCATIONS`/create", "commandStatus": { "status": "SUCCESS", "message": "Stream created", "queryId": null }, "commandSequenceNumber": 2, "warnings": [] } ``` ksqlDB automatically creates the underlying Kafka topic, which is named “quickstart-locations”. Here’s what each parameter in the CREATE STREAM statement does: - `kafka_topic` - Name of the Kafka topic underlying the stream. In this case, it is created automatically, because it doesn’t exist yet, but you can create streams over topics that exist already. - `value_format` - Encoding of the messages stored in the Kafka topic. For JSON encoding, each row is stored as a JSON object whose keys and values are column names and values, for example: ```json {"profileId": "c2309eec", "latitude": 37.7877, "longitude": -122.4205} ``` - `partitions` - Number of partitions to create for the `quickstart-locations` topic. This parameter is not needed for topics that exist already. ## Step 3: Create materialized views The goal of this quick start is to keep track of the latest location of the riders by using a [materialized view](/platform/current/ksqldb/concepts/materialized-views.html). - You create a `currentLocation` table by running a [CREATE TABLE AS SELECT](/platform/current/ksqldb/developer-guide/ksqldb-reference/create-table-as-select.html) statement over the stream you created. The table is updated incrementally as new rider location data arrives. The [LATEST_BY_OFFSET](/platform/current/ksqldb/developer-guide/ksqldb-reference/aggregate-functions.html#latest-by-offset) aggregate function specifies that the application is interested only in the latest location of each rider. - To add more interest, the application materializes a derived table, named `ridersNearMountainView`, that captures the distance of riders from a given location or city. 1. In the query editor, clear the previous query and run the following statement to create the `currentLocation` table. ```sql -- Create the currentLocation table CREATE TABLE currentLocation AS SELECT profileId, LATEST_BY_OFFSET(latitude) AS la, LATEST_BY_OFFSET(longitude) AS lo FROM riderlocations GROUP BY profileId EMIT CHANGES; ``` Your output should resemble: ```json { "@type": "currentStatus", "statementText": "CREATE TABLE CURRENTLOCATION WITH (CLEANUP_POLICY='compact', KAFKA_TOPIC='pksqlc-j3yd58CURRENTLOCATION', PARTITIONS=1, REPLICAS=3, RETENTION_MS=604800000) AS SELECT\n RIDERLOCATIONS.PROFILEID PROFILEID,\n LATEST_BY_OFFSET(RIDERLOCATIONS.LATITUDE) LA,\n LATEST_BY_OFFSET(RIDERLOCATIONS.LONGITUDE) LO\nFROM RIDERLOCATIONS RIDERLOCATIONS\nGROUP BY RIDERLOCATIONS.PROFILEID\nEMIT CHANGES;", "commandId": "table/`CURRENTLOCATION`/create", "commandStatus": { "status": "SUCCESS", "message": "Created query with ID CTAS_CURRENTLOCATION_3", "queryId": "CTAS_CURRENTLOCATION_3" }, "commandSequenceNumber": 4, "warnings": [] } ``` The CREATE TABLE AS SELECT statement created a [persistent query](/platform/current/ksqldb/concepts/queries.html#persistent), named `CTAS_CURRENTLOCATION_3`, that runs continuously on the ksqlDB server. The query’s name is prepended with “CTAS”, which stands for “CREATE TABLE AS SELECT”. 2. In the query editor, clear the previous query and run the following statement to create the `ridersNearMountainView` table. ```sql -- Create the ridersNearMountainView table CREATE TABLE ridersNearMountainView AS SELECT ROUND(GEO_DISTANCE(la, lo, 37.4133, -122.1162), -1) AS distanceInMiles, COLLECT_LIST(profileId) AS riders, COUNT(*) AS count FROM currentLocation GROUP BY ROUND(GEO_DISTANCE(la, lo, 37.4133, -122.1162), -1); ``` Your output should resemble: ```json { "@type": "currentStatus", "statementText": "CREATE TABLE RIDERSNEARMOUNTAINVIEW WITH (CLEANUP_POLICY='compact', KAFKA_TOPIC='pksqlc-j3yd58RIDERSNEARMOUNTAINVIEW', PARTITIONS=1, REPLICAS=3, RETENTION_MS=604800000) AS SELECT\n ROUND(GEO_DISTANCE(CURRENTLOCATION.LA, CURRENTLOCATION.LO, 37.4133, -122.1162), -1) DISTANCEINMILES,\n COLLECT_LIST(CURRENTLOCATION.PROFILEID) RIDERS,\n COUNT(*) COUNT\nFROM CURRENTLOCATION CURRENTLOCATION\nGROUP BY ROUND(GEO_DISTANCE(CURRENTLOCATION.LA, CURRENTLOCATION.LO, 37.4133, -122.1162), -1)\nEMIT CHANGES;", "commandId": "table/`RIDERSNEARMOUNTAINVIEW`/create", "commandStatus": { "status": "SUCCESS", "message": "Created query with ID CTAS_RIDERSNEARMOUNTAINVIEW_5", "queryId": "CTAS_RIDERSNEARMOUNTAINVIEW_5" }, "commandSequenceNumber": 6, "warnings": [] } ``` ## Step 4: Run a push query over the stream In this step, you create a [push query](/platform/current/ksqldb/concepts/queries.html#push) that filters rows in the `riderLocations` stream. This query outputs all rows from the `riderLocations` stream that have coordinates within 5 miles of Mountain View, California. This query never returns, until it’s terminated explicitly. It pushes output rows perpetually to the client as events are written to the `riderLocations` stream. In the query editor, clear the previous query and run the following statement to create a push query that filters rows in the `riderLocations` stream. ```sql -- Mountain View lat, long: 37.4133, -122.1162 SELECT * FROM riderLocations WHERE GEO_DISTANCE(latitude, longitude, 37.4133, -122.1162) <= 5 EMIT CHANGES; ``` With this query, no output is displayed. For now, leave this query running in the current session. In the next step, you write some mock data into the `riderLocations` stream so that the query produces output. ## Step 5: Populate the stream with events Because the session from the previous step is busy waiting for output from the push query, start another session that you can use to write some data into ksqlDB. 1. Press the Ctrl key and click **Editor** to open a new query editor window. 2. In the new editor session, run copy the following INSERT statements and click **Run query**. ```sql INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('c2309eec', 37.7877, -122.4205); INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('18f4ea86', 37.3903, -122.0643); INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('4ab5cbad', 37.3952, -122.0813); INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('8b6eae59', 37.3944, -122.0813); INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('4a7c7b41', 37.4049, -122.0822); INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('4ddad000', 37.7857, -122.4011); ``` No output is shown from the INSERT INTO VALUES statements, but the rows have been entered into the `riderLocations` table. 3. In the browser, click the tab for the previous editor session. The push query that you started in Step 4 now shows the following output: ```json {"PROFILEID":"4a7c7b41","LATITUDE":37.4049,"LONGITUDE":-122.0822} {"PROFILEID":"8b6eae59","LATITUDE":37.3944,"LONGITUDE":-122.0813} {"PROFILEID":"4ab5cbad","LATITUDE":37.3952,"LONGITUDE":-122.0813} ``` Only the rows that have riders who are within 5 miles of Mountain View are displayed. ## Step 6: Run a pull query against the materialized view In this step, you run a [pull query](/platform/current/ksqldb/concepts/queries.html#pull) against the materialized view to retrieve all the riders who are currently within 10 miles of Mountain View. In contrast to the previous push query, which runs continuously, the pull query follows a traditional request-response model and retrieves the latest result from the materialized view. In the second editor session, clear the previous query and run the following statement. ```sql SELECT * from ridersNearMountainView WHERE distanceInMiles <= 10; ``` Your output should resemble: ```json {"DISTANCEINMILES":"10.0","RIDERS":["18f4ea86"],"COUNT":1} {"DISTANCEINMILES":"0.0","RIDERS":["4ab5cbad","8b6eae59","4a7c7b41"],"COUNT":3} ``` ## Step 7: Clean up When you’re done with the quick start, close the second browser tab. 1. Click the browser tab that has the first editor session. 2. Click **Stop query** to end the push query. 3. Click **Persistent queries** to view the two queries that you created in Step 3. 4. For both queries, click **Terminate** and type their names in the delete dialog to confirm deletion. 5. Click **Tables** to view the tables that were created by the queries in Step 3. 6. Click the **RIDERSNEARMOUNTAINVIEW** table to open the details view. 7. Click **DROP TABLE** and type its name in the delete dialog to confirm deletion. The corresponding Kafka topic is also deleted. 8. Repeat the deletion steps for the **CURRENTLOCATION** table. ## Related content - [Pull Queries in Confluent Cloud](pull-queries-in-ccloud.md#ksqldb-pull-queries-in-ccloud) - [Monitor Persistent Queries](monitoring-ksqldb.md#cloud-ksql-monitor) - [Manage Connectors](ksqldb-connector-management.md#cloud-ksql-connector-management) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).