# Quick Start with ksqlDB for Confluent Platform This quick start gets you up and running with ksqlDB for Confluent Platform. The following steps show how to create an Apache Kafka® installation and a ksqlDB Server instance that runs with it. To get started with hosted ksqlDB in Confluent Cloud, see [ksqlDB for Confluent Cloud Quick Start](/cloud/current/ksqldb/ksqldb-quick-start.html). The following steps create a simple streaming application that tracks the latest location of simulated riders: - [Step 1: Create a docker-compose file](#ksqldb-quick-start-create-docker-compose-file) - [Step 2: Start the Kafka broker and ksqlDB server](#ksqldb-quick-start-broker-and-server) - [Step 3: Start the ksqlDB CLI](#ksqldb-quick-start-cli) - [Step 4: Create a stream](#ksqldb-quick-start-create-stream) - [Step 5: Create materialized views](#ksqldb-quick-start-create-materialized-views) - [Step 6: Run a push query over the stream](#ksqldb-quick-start-run-push-query) - [Step 7: Populate the stream with events](#ksqldb-quick-start-populate-stream) - [Step 8: Run a pull query against the materialized view](#ksqldb-quick-start-run-pull-query) - [Step 9: Clean up](#ksqldb-quick-start-clean-up) ## Prerequisites - Docker and Docker Compose installed on a computer with a supported [Operating System](../installation/versions-interoperability.md#operating-systems). - Make sure Docker is running. - For full prerequisites, expand **Detailed prerequisites** in the following section. ### **Detailed prerequisites** - A connection to the internet. - [Operating System](../installation/versions-interoperability.md#operating-systems) currently supported by Confluent Platform. #### NOTE You can run the ksqlDB Quick Start on Windows if you are running Docker Desktop for Windows on WSL 2. For more information, see [How to Run Confluent on Windows in Minutes](https://www.confluent.io/blog/set-up-and-run-kafka-on-windows-and-wsl-2/). - Docker version 1.11 or later is [installed and running](https://docs.docker.com/engine/installation/). - **On Mac:** Docker memory is allocated minimally at 6 GB (Mac). When using Docker Desktop for Mac, the default Docker memory allocation is 2 GB. Change the default allocation to 6 GB in the Docker Desktop app by navigating to **Preferences** > **Resources** > **Advanced**. - Networking and Kafka on Docker: Configure your hosts and ports to allow both internal and external components to the Docker network to communicate. For more information, see [Kafka Listeners Explained](https://www.confluent.io/blog/kafka-listeners-explained/). ## Step 1: Create a docker-compose file The minimum set of services for running ksqlDB comprises a Kafka broker and ksqlDB Server. The ksqlDB CLI is required for developing applications with SQL code. The following `docker-compose` file specifies the Docker images that you need for a minimal local environment: - confluentinc/cp-kafka - confluentinc/cp-ksqldb-server - confluentinc/cp-ksqldb-cli 1. Run the following command to create a file named `docker-compose.yml`. ```bash touch docker-compose.yml ``` 2. Copy the following YAML into docker-compose.yml and save the file. ```yaml --- version: '2' services: broker: image: confluentinc/cp-kafka:8.3.1 hostname: broker container_name: broker ports: - "9092:9092" - "9101:9101" environment: KAFKA_NODE_ID: 1 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT' KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092' KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_JMX_PORT: 9101 KAFKA_JMX_HOSTNAME: localhost KAFKA_PROCESS_ROLES: 'broker,controller' KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093' KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092' KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT' KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs' # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid" # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qk' ksqldb-server: image: confluentinc/cp-ksqldb-server:8.3.1 hostname: ksqldb-server container_name: ksqldb-server depends_on: - broker ports: - "8088:8088" environment: KSQL_CONFIG_DIR: "/etc/ksql" KSQL_BOOTSTRAP_SERVERS: "broker:29092" KSQL_HOST_NAME: ksqldb-server KSQL_LISTENERS: "http://0.0.0.0:8088" KSQL_CACHE_MAX_BYTES_BUFFERING: 0 KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081" KSQL_KSQL_CONNECT_URL: "http://connect:8083" KSQL_KSQL_LOGGING_PROCESSING_TOPIC_REPLICATION_FACTOR: 1 KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: 'true' KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: 'true' ksqldb-cli: image: confluentinc/cp-ksqldb-server:8.1.0 container_name: ksqldb-cli depends_on: - broker - ksqldb-server entrypoint: /bin/sh tty: true ``` ## Step 2: Start the Kafka broker and ksqlDB server In the directory where you saved the `docker-compose.yml` file, run the following command to start the Kafka broker and ksqlDB Server. ```bash docker-compose up ``` The `docker-compose` tool starts downloading the Docker images. Downloading may take a few minutes. Once the images are downloaded, the containers are created. Your output should resemble: ```none [+] Running 4/4 ✔ Network ksqldb-quick-start_default Created 0.0s ✔ Container broker Created 0.1s ✔ Container ksqldb-server Created 0.1s ✔ Container ksqldb-cli Created ``` The Kafka broker and the ksqlDB Server emit numerous status messages as they start. This behavior is normal. ksqlDB runs as a server which clients connect to in order to issue queries. ## Step 3: Start the ksqlDB CLI To submit SQL statements to the ksqlDB Server, start the ksqlDB CLI. 1. Open a new command shell. 2. Run the following command to start the ksqlDB CLI container. ```bash docker exec -it ksqldb-cli ksql http://ksqldb-server:8088 ``` Your output should resemble: ```text =========================================== = _ _ ____ ____ = = | | _____ __ _| | _ \| __ ) = = | |/ / __|/ _` | | | | | _ \ = = | <\__ \ (_| | | |_| | |_) | = = |_|\_\___/\__, |_|____/|____/ = = |_| = = The Database purpose-built = = for stream processing apps = =========================================== Copyright 2017-2022 Confluent Inc. CLI v8.3.1, Server v8.3.1 located at http://localhost:8088 Server Status: RUNNING Having trouble? Type 'help' (case-insensitive) for a rundown of how things work! ksql> ``` The `ksql>` prompt is where you enter SQL statements that run on the ksqlDB Server. ## Step 4: Create a stream You’re ready to create a [stream](concepts/streams.md#ksqldb-concepts-streams). A stream associates a schema with an underlying Kafka topic. You use the [CREATE STREAM](developer-guide/ksqldb-reference/create-stream.md#ksqldb-reference-create-stream) statement to register a stream on a topic. If the topic doesn’t exist yet, ksqlDB creates it on the Kafka broker. In the ksqlDB CLI, copy the following SQL and press Enter to run the statement. ```sql CREATE STREAM riderLocations (profileId VARCHAR, latitude DOUBLE, longitude DOUBLE) WITH (kafka_topic='locations', value_format='json', partitions=1); ``` Your output should resemble: ```none Message ---------------- Stream created ---------------- ``` 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’s 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 `locations` topic. This parameter is not needed for topics that exist already. ## Step 5: 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](concepts/materialized-views.md#ksqldb-concepts-materialized-views). - You create a `currentLocation` [table](concepts/tables.md#ksqldb-concepts-tables) by running a [CREATE TABLE AS SELECT](developer-guide/ksqldb-reference/create-table-as-select.md#ksqldb-reference-create-table-as-select) statement over the previously created stream. The table is updated incrementally as new rider location data arrives. The [LATEST_BY_OFFSET](developer-guide/ksqldb-reference/aggregate-functions.md#ksqldb-reference-aggregate-functions-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 ksqlDB CLI, 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: ```none Message ---------------------------------------------- Created query with ID CTAS_CURRENTLOCATION_3 ---------------------------------------------- ``` The CREATE TABLE AS SELECT statement created a [persistent query](concepts/queries.md#ksqldb-concepts-queries-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. 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: ```none Message ----------------------------------------------------- Created query with ID CTAS_RIDERSNEARMOUNTAINVIEW_5 ----------------------------------------------------- ``` ## Step 6: Run a push query over the stream In this step, you create a [push query](developer-guide/ksqldb-reference/select-push-query.md#ksqldb-reference-select-push-query) 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 ksqlDB CLI, 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; ``` Your output should resemble: ```none +--------------------+-------------------+--------------------+ |PROFILEID |LATITUDE |LONGITUDE | +--------------------+-------------------+--------------------+ Press CTRL-C to interrupt ``` For now, leave this query running in the CLI session. In the next step, you write some mock data into the `riderLocations` stream so that the query produces output. ## Step 7: Populate the stream with events Because the CLI 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. Open a new command shell. 2. Run the following command to start another ksqlDB CLI instance. ```bash docker exec -it ksqldb-cli ksql http://ksqldb-server:8088 ``` 3. In the new CLI session, run the following INSERT statements. ```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); ``` 4. Open the previous command shell to view the output from the push query that you started in [Step 6](#ksqldb-quick-start-run-push-query). Your output should resemble: ```none +--------------------+-------------------+--------------------+ |PROFILEID |LATITUDE |LONGITUDE | +--------------------+-------------------+--------------------+ |4ab5cbad |37.3952 |-122.0813 | |8b6eae59 |37.3944 |-122.0813 | |4a7c7b41 |37.4049 |-122.0822 | Press CTRL-C to interrupt ``` Only the rows that have riders who are within 5 miles of Mountain View are displayed. ## Step 8: Run a pull query against the materialized view In this step, you run a [pull query](developer-guide/ksqldb-reference/select-pull-query.md#ksqldb-reference-select-pull-query) against the materialized view to retrieve all of 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 ksqlDB CLI, run the following statement. ```sql SELECT * from ridersNearMountainView WHERE distanceInMiles <= 10; ``` Your output should resemble: ```none +--------------------+-------------------------------+------+ |DISTANCEINMILES |RIDERS |COUNT | +--------------------+-------------------------------+------+ |0.0 |[4ab5cbad, 8b6eae59, 4a7c7b41] |3 | |10.0 |[18f4ea86] |1 | Query terminated ``` ## Step 9: Clean up When you’re done with the quick start, close the ksqlDB CLI instances and stop the ksqlDB Server and Kafka broker. 1. In the second ksqlDB CLI instance, type “exit” and press enter. Your output should resemble: ```none Exiting ksqlDB. ``` 2. In the ksqlDB CLI instance that’s running the push query, press Ctrl-C to end the push query. Your output should resemble: ```none ^CQuery terminated ``` 3. Type “exit” and press enter to close the first ksqlDB CLI instance. 4. Run the following command to stop the running containers. ```bash docker-compose down ``` Your output should resemble: ```none [+] Running 4/4 ✔ Container ksqldb-cli Removed ✔ Container ksqldb-server Removed ✔ Container broker Removed ✔ Network test-ksqldb-quick-start_default Removed ``` ## Related content - [ksqlDB for Confluent Cloud Quick Start](/cloud/current/ksqldb/ksqldb-quick-start.html)