Contact UsTry it for free
© 2026 Direct Cursus Technology L.L.C.

Delivering data to ksqlDB

Written by
Updated at February 6, 2026

ksqlDB is a database designed for stream processing of messages coming from Apache Kafka® topics. Working with message streams in ksqlDB is similar to working with tables in a regular database. A ksqlDB table is automatically populated with data from an Apache Kafka® topic, and any data you insert into the ksqlDB table is written back to the topic. Learn more in this ksqlDB article.

To set up data delivery from Managed Service for Apache Kafka® to ksqlDB:

  1. Set up Apache Kafka® integration for the ksqlDB database.
  2. Review the format of the data coming from Managed Service for Apache Kafka®.
  3. Create a ksqlDB table to write a data stream from the Apache Kafka® topic.
  4. Get test data from the Managed Service for Apache Kafka® cluster.
  5. Write the test data to ksqlDB.
  6. Check the Apache Kafka® topic for the test data.

If you no longer need the resources you created, delete them.

The support cost for this solution includes:

  • Managed Service for Apache Kafka® cluster fee, which covers the use of computing resources allocated to hosts (including ZooKeeper hosts) and disk space (see Apache Kafka® pricing).
  • Fee for public IP addresses if public access is enabled for cluster hosts (see Virtual Private Cloud pricing).

Getting started

  1. Create a Managed Service for Apache Kafka® cluster in any suitable configuration.

    • If the ksqlDB server is hosted on the internet, create a publicly accessible Managed Service for Apache Kafka® cluster.

    • If the ksqlDB server is hosted in Yandex Cloud, create a Managed Service for Apache Kafka® cluster on the same cloud network as ksqlDB.

  2. Create topics in the Managed Service for Apache Kafka® cluster:

    1. Service topic named _confluent-ksql-default__command_topic configured as follows:
      • Replication factor: 1
      • Number of partitions: 1
      • Log cleanup policy: Delete
      • Log segment lifetime, ms: -1
      • Minimum number of in-sync replicas: 1
    2. Service topic named default_ksql_processing_log to write ksqlDB logs to. Use any settings.
    3. Data storage topic named locations. Use any settings.

  3. Create a user named ksql and assign them the ACCESS_ROLE_ADMIN role for all topics.

  4. Make sure you can connect to the ksqlDB server.

  5. Install kafkacat on the ksqlDB server and check that you can use it to connect to the Managed Service for Apache Kafka® cluster over SSL.

  6. Install jq on the ksqlDB server to process JSON data streams.

Set up Apache Kafka® integration for the ksqlDB database

  1. Connect to the ksqlDB server.

  2. Add the server's SSL certificate to the Java trusted certificate store (Java Key Store) so that ksqlDB can use it for secure connections to cluster hosts. Set a password in the -storepass parameter for additional storage protection:

    cd /etc/ksqldb && \
sudo keytool -importcert -alias YandexCA -file /usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt \
-keystore ssl -storepass <certificate_store_password> \
--noprompt

  3. In the /etc/ksqldb/ksql-server.properties ksqlDB configuration file, specify the credentials for authentication in the Managed Service for Apache Kafka® cluster:

    bootstrap.servers=<broker_1_FQDN>:9091,...,<broker_N_FQDN>:9091
sasl.mechanism=SCRAM-SHA-512
security.protocol=SASL_SSL
ssl.truststore.location=/etc/ksqldb/ssl
ssl.truststore.password=<certificate_store_password>
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="ksql" password="<ksql_user_password>";

    To learn how to get a broker host FQDN, see this guide.

    You can get the cluster name with the list of clusters in the folder.

  4. In the ksqlDB logging configuration file named /etc/ksqldb/log4j.properties, configure logging to a Managed Service for Apache Kafka® cluster topic:

    log4j.appender.kafka_appender=org.apache.kafka.log4jappender.KafkaLog4jAppender
log4j.appender.kafka_appender.layout=io.confluent.common.logging.log4j.StructuredJsonLayout
log4j.appender.kafka_appender.BrokerList=<broker_1_FQDN>:9091,...,<broker_N_FQDN>:9091
log4j.appender.kafka_appender.Topic=default_ksql_processing_log
log4j.logger.io.confluent.ksql=INFO,kafka_appender

log4j.appender.kafka_appender.clientJaasConf=org.apache.kafka.common.security.scram.ScramLoginModule required username="ksql" password="<ksql_user_password>";
log4j.appender.kafka_appender.SecurityProtocol=SASL_SSL
log4j.appender.kafka_appender.SaslMechanism=SCRAM-SHA-512
log4j.appender.kafka_appender.SslTruststoreLocation=/etc/ksqldb/ssl
log4j.appender.kafka_appender.SslTruststorePassword=<certificate_store_password>

  5. Restart the ksqlDB service with the command below:

    sudo systemctl restart confluent-ksqldb.service

Review the format of the data coming from Managed Service for Apache Kafka®

The Managed Service for Apache Kafka® data stream processing depends on the Apache Kafka® message view format.

In our example, geodata is written to the locations Apache Kafka® topic in JSON format:

  • profileId: ID
  • latitude: Latitude
  • longitude: Longitude

This data will be transmitted as Apache Kafka® messages. Each message will contain a string containing a serialized JSON object with the following structure:

{"profileId": "c2309eec", "latitude": 37.7877, "longitude": -122.4205}

ksqlDB stores the values of the relevant parameters from Apache Kafka® messages in a three-column table.

Next, we are going to configure the fields of a ksqlDB data stream table.

Create a table in ksqlDB to write a data stream from the Apache Kafka® topic

Create a table in ksqlDB for writing data from the Apache Kafka® topic. The table structure matches the format of data from Managed Service for Apache Kafka®:

  1. Connect to the ksqlDB server.

  2. Run the ksql client using this command:

    ksql http://0.0.0.0:8088

  3. Run this query:

    CREATE STREAM riderLocations 
(
  profileId VARCHAR,
  latitude DOUBLE,
  longitude DOUBLE
) WITH 
(
  kafka_topic='locations', 
  value_format='json', 
  partitions=<number_of_locations_topic_sections>
);

    This data stream table will be automatically populated with messages from the Managed Service for Apache Kafka® cluster's locations topic. ksqlDB uses the ksql user's settings to read messages.

    For more information about creating a streaming table in ksqlDB, see this ksqlDB article.

  4. Run this query:

    SELECT * FROM riderLocations WHERE 
         GEO_DISTANCE(latitude, longitude, 37.4133, -122.1162) <= 5 
         EMIT CHANGES;

    The query waits for real-time data to appear in the table.

Get test data from the Managed Service for Apache Kafka® cluster

  1. Connect to the ksqlDB server.

  2. Create a file named sample.json with the following test data:

    {
  "profileId": "c2309eec", 
  "latitude": 37.7877,
  "longitude": -122.4205
}

{
  "profileId": "4ab5cbad", 
  "latitude": 37.3952,
  "longitude": -122.0813
}

{
  "profileId": "4a7c7b41", 
  "latitude": 37.4049,
  "longitude": -122.0822
}

  3. Send the sample.json file to the Managed Service for Apache Kafka® cluster's locations topic using jq and kafkacat:

    jq -rc . sample.json | kafkacat -P \
   -b <broker_1_FQDN>:9091,...,<broker_N_FQDN>:9091> \
   -t locations \
   -X security.protocol=SASL_SSL \
   -X sasl.mechanisms=SCRAM-SHA-512 \
   -X sasl.username=ksql \
   -X sasl.password="<ksql_user_password>" \
   -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt -Z

    The information is sent using the ksql user. To learn more about setting up an SSL certificate and using kafkacat, see Connecting to an Apache Kafka® cluster from applications.

  4. Make sure the session shows the data sent to the topic:

    +--------------------------+--------------------------+------------------------+
|PROFILEID                 |LATITUDE                  |LONGITUDE               |
+--------------------------+--------------------------+------------------------+
|4ab5cbad                  |37.3952                   |-122.0813               | 
|4a7c7b41                  |37.4049                   |-122.0822               |

The data is read using the ksql user.

Write test data to ksqlDB

  1. Connect to the ksqlDB server.

  2. Run the ksql client using this command:

    ksql http://0.0.0.0:8088

  3. Insert test data into the riderLocations table as follows:

    INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('18f4ea86', 37.3903, -122.0643);
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('8b6eae59', 37.3944, -122.0813);
INSERT INTO riderLocations (profileId, latitude, longitude) VALUES ('4ddad000', 37.7857, -122.4011);

    This data is sent synchronously to the locations Apache Kafka® topic using the ksql user.

Check for records in Apache Kafka® topics

  1. Check messages in the Managed Service for Apache Kafka® cluster's locations topic using kafkacat and the ksql user:

    kafkacat -C \
 -b <broker_1_FQDN>:9091,...,<broker_N_FQDN>:9091 \
 -t locations \
 -X security.protocol=SASL_SSL \
 -X sasl.mechanisms=SCRAM-SHA-512 \
 -X sasl.username=ksql \
 -X sasl.password="<ksql_user_password>" \
 -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt -Z -K:

  2. Make sure the console displays the messages you inserted into the table.

  3. Check messages in the Managed Service for Apache Kafka® cluster's default_ksql_processing_log topic using kafkacat and the ksql user:

    kafkacat -C \
 -b <broker_1_FQDN>:9091,...,<broker_N_FQDN>:9091 \
 -t default_ksql_processing_log \
 -X security.protocol=SASL_SSL \
 -X sasl.mechanisms=SCRAM-SHA-512 \
 -X sasl.username=ksql \
 -X sasl.password="<ksql_user_password>" \
 -X ssl.ca.location=/usr/local/share/ca-certificates/Yandex/YandexInternalRootCA.crt -Z -K:

  4. Make sure the console displays ksqlDB log entries.

Delete the resources you created

Some resources are not free of charge. Delete the resources you no longer need to avoid paying for them:

Previous
Delivering data to ClickHouse®
Next
Synchronizing Apache Kafka® topics in Object Storage without using the internet