ksqlDB

The ksqlDB section is where you run KSQL against a ksqlDB cluster and look at what it holds: its streams, tables, persistent queries and topics, and the connectors it manages. KafkIO talks to ksqlDB's REST API for all of this, so what you can do is what that ksqlDB server allows.

To open it, connect the cluster and click ksqlDB under its name in the Kafka Clusters pane. The entry is greyed out until at least one ksqlDB cluster is set up in the cluster's settings (in the screenshot below, Storefront Prod (eu-west-1) has none). The section opens on the KSQL Editor.

1 2 3 4 5 6 7 8 9 10 11 12 13

1. ksqlDB cluster: which ksqlDB cluster every tab works with (below).
2. Refresh ksqlDB Cluster List.
3. The tabs: KSQL Editor, then the lists (Streams, Tables, Persistent Queries, Topics and Connectors).
4. KSQL: the editor, with syntax highlighting and a Search KSQL... box above it.
5. Examples (below).
6. Execute KSQL, which reads Execute selected while text is highlighted. Greyed out here because a push query is running.
7. Stream Properties (below).
8. The Default stream properties example.
9. The results tabs, JSON and Table (below).
10. Filter results...
11. How many rows the results hold.
12. Stop Stream, while a push query runs (below).
13. Duplicate Tab: open the section in a tab of its own, as was done here for Topics (Duplicating a tab).

Choosing a ksqlDB cluster

The section works with one ksqlDB cluster at a time: the one in the ksqlDB cluster list 1, which has one entry per ksqlDB configuration in the cluster's settings, by the name you gave it. With only one, there's nothing to choose. Every tab, including the editor, belongs to the ksqlDB cluster selected when you use it, and KafkIO keeps a separate set for each: switch to another ksqlDB cluster and you get its own editor and lists; switch back and your KSQL, results and fetched lists are as you left them. To work with two ksqlDB clusters side by side, duplicate the tab 13 and pick a different one in each.

Refresh ksqlDB Cluster List 2 re-reads the list from the cluster's settings; it doesn't contact ksqlDB. KafkIO only reaches ksqlDB when you run something or fetch a list, so a ksqlDB server that's down or misconfigured shows up then, as an error. Setting up, naming and testing ksqlDB connections is in ksqlDB Connections.

The KSQL Editor

Type KSQL into the KSQL box 4, or paste it. It opens with two sample queries on riderlocations (the stream from ksqlDB's own quick start); replace them with your own. You can keep several statements in the box as a scratch pad, but only one runs at a time.

Running a statement

Click Execute KSQL 6 to run what's in the box. If it holds more than one statement, highlight the one you want first: the button then reads Execute selected and runs only the highlighted text. With several statements and nothing highlighted, KafkIO refuses before anything is sent:

KafkIO also checks the statement's syntax before sending it ("Sql statement is invalid or unsupported" means it couldn't parse it), and doesn't run PRINT, DEFINE or UNDEFINE ("Unsupported KSQL statement type"). Everything else goes to ksqlDB, and what happens next depends on the kind of statement:

KafkIO looks for EMIT as a keyword, so EMIT CHANGES inside a comment or a string doesn't turn a pull query into a push query. A CREATE STREAM ... AS SELECT ... EMIT CHANGES is a statement, not a push query: ksqlDB starts a persistent query for it and answers straight away.

Built-in examples

The Examples links 5 under the editor add KSQL to the end of the box (they don't replace what's there), to adapt or to remind you of the syntax:

• Select: a pull query, a push query using a function (UCASE), and a filtered push query.
• Insert: an INSERT INTO ... VALUES.
• Create: a CREATE STREAM over a topic and a CREATE TABLE ... AS SELECT with an aggregation.
• Adhoc: a cheat sheet of SHOW, DESCRIBE, DROP, PAUSE/RESUME, TERMINATE, EXPLAIN and CREATE SINK CONNECTOR statements, with placeholders such as my_stream_name and <query_id> to fill in.

The examples work on the quick-start's riderLocations stream, which your ksqlDB server probably doesn't have: change the names before you run them.

Stream properties

The Stream Properties box 7 holds a JSON object of properties sent with every statement and query you run (as ksqlDB's streamsProperties). It starts as, and Default 8 puts back:

{
   "auto.offset.reset": "latest"
}

With latest, a push query only shows rows written after it starts, and a CREATE ... AS SELECT only processes new data. Change it to earliest (as in the screenshot) to read a stream from the beginning. Add any other property ksqlDB accepts per request, such as "ksql.streams.cache.max.bytes.buffering": 0. JSON validation under the box says success once the text is valid JSON; when it isn't, running anything fails before it reaches ksqlDB. Clear the box to send no properties.

The properties belong to the editor of the selected ksqlDB cluster and are kept while KafkIO runs, not between sessions.

LIMIT handling

So a pull query can't fetch a whole stream into KafkIO by accident, KafkIO caps it at 10,000 rows: a pull query without a LIMIT gets LIMIT 10000 added, and a larger LIMIT is lowered to 10,000. When it does this, a warning shows across the top of the cluster tab for ten seconds:

If you need more than 10,000 rows, query in slices (by key or time range), or write the result to a topic with CREATE STREAM ... AS SELECT and read the topic instead. Push queries get no LIMIT added (see below for how many rows they keep).

Reading the results

Results appear below the editor, which you can give more room by dragging the divider, or undock into a window of their own with the arrow at their top right (Split panes and detail tabs). Two tabs run down their left side 9:

• Table: one column per column of the result, named as ksqlDB names it (in capitals unless quoted). Structs, maps and arrays show as JSON in small scrollable boxes. Click a header to sort, type in Filter results... 10 to keep only the rows that contain some text (or a regular expression, with the .* toggle), and read the row count in the corner 11. Newer rows are added at the top. Right-click a row to copy it (Copying to the clipboard).
• JSON: for statements, ksqlDB's whole response, pretty-printed, with a Search results... box. It's the most useful view for DESCRIBE, EXPLAIN and SHOW PROPERTIES, whose answers are nested. It's greyed out for queries, whose rows only appear in the table.

1 2

1. JSON.
2. Search results..., with arrows to step through the matches.

When ksqlDB rejects something, how you find out depends on what you ran:

• A query (pull or push) that fails, for example on a stream that doesn't exist, opens a ksqlDB Query Error dialog with ksqlDB's message ("RETURNS_STREAM does not exist", with its error code), and the Table tab shows the time, the KSQL that was sent (with any LIMIT KafkIO added) and the full error.
• A statement that fails shows no dialog: ksqlDB's answer is the result, a row with @type statement_error, its error_code and message (for example "Stream RETURNS_STREAM does not exist."). Check the results after a DROP or CREATE rather than assuming it worked.

A request ksqlDB doesn't answer in time fails with a timeout; the ksqlDB timeouts are under Kafka & Timeouts.

Push queries

A push query (EMIT CHANGES or EMIT FINAL) keeps running on the ksqlDB server and sends each new row to KafkIO as it's produced. While it runs:

• A flashing Stop Stream button 12 sits at the bottom right of the results, and Execute KSQL is greyed out: one push query at a time per editor. Click Stop Stream to end the query; KafkIO terminates it on the server (so it doesn't keep using ksqlDB's resources) and the rows stay in the table. A query with its own LIMIT stops by itself once it has sent that many rows.
• The arrow beside Stop Stream opens a menu with Notification sounds: tick it for a sound on each new row, handy for a filtered query where rows are rare, such as WHERE status = 'FAILED'. It's off for every new push query. Sounds work as for topic streams (Sound notifications).
• The table keeps the newest 10,000 rows: once there are more, the oldest are dropped, 500 at a time, so a long-running query doesn't use ever more memory. KafkIO also takes rows from ksqlDB at up to 40 a second, so a fast stream falls behind; filter the query to the rows you need, or write it to a topic with CREATE STREAM ... AS SELECT and look at that.

What a push query starts from is set by auto.offset.reset in the stream properties: with the default latest, an empty table just means nothing new has arrived yet. Closing the cluster tab stops a running push query.

Lists load when you ask

The other five tabs list what the selected ksqlDB cluster holds. Each one asks ksqlDB only when you want it to: the first time you open a tab it shows "No streams to display..." (or tables, queries and so on) and a Fetch button 1. That keeps the section quick to open on a busy ksqlDB server, and avoids asking it for lists you don't need.

1

After that, each tab keeps what it read until you refresh it with the button at the end of its filter box, so streams, tables or queries you create in the editor appear once you refresh. Every tab has a filter box (Filtering), sortable columns, a count in the corner, and a right-click menu to copy a value, a row (Copy Summary) or the whole list (Copy All Summaries).

Streams and tables

The Streams and Tables tabs list ksqlDB's streams and tables (what SHOW STREAMS and SHOW TABLES return), with the same columns.

1 2 3 4

1. Streams.
2. Filter streams...: matches the name, topic and formats.
3. Refresh All Streams.
4. Windowed: Yes for windowed tables (from a WINDOW TUMBLING, HOPPING or SESSION aggregation), whose keys include the window.

Name is the stream or table's name in KSQL, in capitals unless it was created with a quoted name. Topic is the Kafka topic behind it, which you can open in the Topics section; Key Format and Value Format say how its keys and values are serialized (KAFKA, JSON, AVRO, PROTOBUF, ...), which is also what to pick when you read that topic in KafkIO. KSQL_PROCESSING_LOG is ksqlDB's own stream of processing errors (records it couldn't deserialize, for example): query it when a stream or table is missing rows you expected.

To see a stream or table's columns, run DESCRIBE name; (or DESCRIBE name EXTENDED;) in the editor and read the JSON tab.

Persistent queries

The Persistent Queries tab lists the queries ksqlDB runs continuously: one for each CREATE STREAM ... AS SELECT, CREATE TABLE ... AS SELECT and INSERT INTO ... SELECT.

1 2 3

1. Persistent Queries.
2. SQL: the full statement as ksqlDB stored it, with every default filled in, in a box you can scroll and copy from.
3. Sink Topic: the Kafka topic the query writes to.

The ID (such as CTAS_SPEND_BY_CUSTOMER_3) is what ksqlDB's own commands take. The tab has no buttons to stop or pause a query: copy the ID and run TERMINATE ID;, PAUSE ID; or RESUME ID; in the editor (they're in the Adhoc example), then refresh the tab. EXPLAIN ID; shows the query's plan and status on the JSON tab. ksqlDB won't drop a stream or table while a query writes to it: terminate the query first.

Topics

The Topics tab lists the Kafka topics ksqlDB can see (as SHOW TOPICS does), with their Partitions and Total Replicas (the replicas of all partitions added up: a 3-partition topic with a replication factor of 3 shows 9). Use it to check a topic's name and partition count before a CREATE STREAM ... WITH (KAFKA_TOPIC=...), and to see that ksqlDB, whose permissions may differ from yours, can see it. For everything else about a topic, use the Topics section.

Connectors

ksqlDB can manage Kafka Connect connectors with CREATE SOURCE|SINK CONNECTOR, SHOW CONNECTORS and DROP CONNECTOR, either on a Connect cluster it's linked to (ksql.connect.url) or on Connect workers embedded in the ksqlDB server. The Connectors tab does these three things for you. If the ksqlDB server has no Connect behind it, fetching the list fails.

1 2 3 4 5 6

1. Connectors.
2. The tab's own note: it manages connectors only as far as ksqlDB can.
3. Create Connector (below).
4. Delete Connector, for the selected connector (below).
5. Refresh All Connectors.
6. State: the connector's state and how many of its tasks are running.

Each row shows the connector's Name, Type (SOURCE or SINK), Classname (its connector.class), State, and the Sources and Topics ksqlDB associates with it (often empty: ksqlDB fills them only for connectors whose topics it knows). Connectors created outside ksqlDB on the same Connect cluster are listed too, like orders-archive-sink here.

Creating a connector through ksqlDB

Click Create Connector 3. KafkIO sends a CREATE SOURCE CONNECTOR or CREATE SINK CONNECTOR statement built from the dialog to the selected ksqlDB cluster, so you don't have to write the WITH (...) clause by hand.

See below 1 2 3 4 5 6 7

1. Name: the connector's name on Connect (required). Names with - or other characters KSQL doesn't allow in identifiers must be wrapped in back-quotes (see the warning below).
2. Type: Source (the default) or Sink. It must match the connector class.
3. Overwrite if exists: doesn't do what it says; see below.
4. The connector's configuration, as a JSON object of its properties (everything that would go in the WITH (...) clause). The name comes from the field above, so leave name out.
5. Examples: each link replaces the JSON with a template for that connector class, to fill in. The connector's plugin must be installed on the Connect workers.
6. JSON validation: success once the text is valid JSON.
7. Create Connector, greyed out until there's a name.

On success KafkIO says so in a small dialog, the dialog stays open (to create another), and the list refreshes. Close it with Close. A connector that's created but can't start (a missing plugin setting, an unreachable database) still appears, with a state such as FAILED: the reason is in the Connect worker's log, or in the Kafka Connect section if you have it.

If the connector already exists

ksqlDB can't change an existing connector's configuration (there's no CREATE OR REPLACE CONNECTOR), and in the current version the Overwrite if exists box works the opposite way to its label and hint:

• Ticked (the default): creating a connector whose name is taken fails with "Connector ... already exists" (409). Nothing is overwritten.
• Unticked: KafkIO sends CREATE ... CONNECTOR IF NOT EXISTS, so a taken name reports success but the existing connector keeps its old configuration.

To change a connector's configuration, drop it and create it again, or edit it in the Kafka Connect section.

Dropping a connector

Select a connector and click Delete Connector 4 (or right-click it and choose Delete Connector), then confirm with Delete in the Delete Connector ... dialog. KafkIO runs DROP CONNECTOR, which deletes the connector from Connect for everyone, whichever way it was created, and refreshes the list. It doesn't delete its topics, the consumer group of a sink (connect-NAME) or the stored offsets of a source.

Related

• ksqlDB Connections: adding ksqlDB clusters to a Kafka cluster, and testing them
• Kafka Connect: full connector management
• Streaming Messages Live: watching a topic without KSQL
• Kafka & Timeouts: ksqlDB timeouts
• Duplicating a tab