SELECT (Push Query)
1 2 3 4 5 6 7 8 9 10 11 12
Push a continuous stream of updates to the ksqlDB stream or table. The result of this statement isn't persisted in a Kafka topic and is printed out only in the console, or returned to the client. To stop a push query started in the CLI press Ctrl+C.
Execute a push query via the CLI or by sending an HTTP request to the ksqlDB REST API, and the API sends back a chunked response of indefinite length.
Push queries enable you to subscribe to changes, which enable reacting to new information in real-time. They’re a good fit for asynchronous application flows. For request/response flows, see Pull Queries.
See push queries in action
Push queries can use all available SQL features, which can be useful when prototyping a persistent query or when running ad-hoc queries from the CLI. But unlike persistent queries, push queries are not shared. If multiple clients submit the same push query, ksqlDB computes independent results for each client.
If you're using push queries from an application, move all the heavy lifting into a persistent query and keep your push query as simple as possible.
In the previous statements,
from_item is one of the following:
stream_name [ alias ]
table_name [ alias ]
from_item LEFT JOIN from_item ON join_condition
The WHERE clause can refer to any column defined for a stream or table,
ROWOFFSET pseudo columns.
where_condition is an expression that evaluates to true for each record selected.
In the WHERE expression, you can use any operator that ksqlDB supports. For more information, see Operators in ksqlDB.
The EMIT clause lets you control the output refinement of your push query. The output refinement is how you would like to emit your results.
ksqlDB supports the following output refinement types.
This is the standard output refinement for push queries, for when we would like to see all changes happening.
The following statement shows how to select all records from a
stream that have timestamps between two values.
1 2 3 4
When writing logical expressions using
ROWTIME, you can use ISO-8601
formatted date strings to represent date times. For example, the previous
query is equivalent to the following:
1 2 3 4
If the datestring is inexact, the rest of the timestamp is assumed to be
padded with 0s. For example,
ROWTIME = '2019-07-30T11:00' is
ROWTIME = '2019-07-30T11:00:00.0000'.
You can specify time zones within the datestring. For example,
2017-11-17T04:53:45-0330 is in the Newfoundland time zone. If no
timezone is specified within the datestring, then timestamps are
interpreted in the UTC time zone.
You use the
LIMIT clause to limit the number of rows returned. Once the
limit is reached, the query terminates.
The following statement shows how to select five records from a
If no limit is supplied the query runs until terminated, streaming back all results to the console.
If you want to select older data, you can configure ksqlDB to query the stream from the beginning. You must run this configuration before running the query:
You can output a struct from a query
by using a SELECT statement. The following example creates a struct from a
You can use the WINDOW clause only if the
from_item is a stream.
The WINDOW clause lets you control how to group input records that have the same key into so-called windows for operations like aggregations or joins. Windows are tracked per record key. For more information, see Time and Windows in ksqlDB.
Windowing adds two additional system columns to the data, which provide
the window bounds:
ksqlDB supports the following WINDOW types.
Tumbling windows group input records into fixed-sized, non-overlapping windows based on the records' timestamps. You must specify the window size for tumbling windows. Tumbling windows are a special case of hopping windows, where the window size is equal to the advance interval.
The following statement shows how to create a push query that has a tumbling window.
1 2 3 4 5
Hopping windows group input records into fixed-sized, (possibly) overlapping windows based on the records' timestamps. You must specify the window size and the advance interval for hopping windows.
The following statement shows how to create a push query that has a hopping window.
1 2 3 4 5
Session windows group input records into so-called sessions. You must specify the session inactivity gap parameter for session windows. For example, imagine you set the inactivity gap to 5 minutes. If, for a given record key such as "alice", no new input data arrives for more than 5 minutes, then the current session for "alice" is closed, and any newly arriving data for "alice" in the future will mark the beginning of a new session.
The following statement shows how to create a push query that has a session window.
1 2 3 4 5
WITHIN and GRACE PERIOD¶
Stream-Stream joins must have a WITHIN clause specified.
The WITHIN clause lets you specify a time range in a windowed join. When you join two streams, you must specify a WITHIN clause for matching records that both occur within a specified time interval.
Here's an example stream-stream-stream join that combines
streams. The resulting
shipped_orders stream contains all orders paid within 1 hour of when
the order was placed, and shipped within 2 hours of the payment being received.
1 2 3 4 5 6 7 8 9
The GRACE PERIOD, part of the WITHIN clause, allows the join to process out-of-order records for up to the specified grace period. Events that arrive after the grace period has passed are dropped as late records and not joined.
1 2 3 4 5 6 7 8 9
If you don't specify a grace period explicitly, the default grace period is 24 hours. This could cause a huge amount of disk usage on high-throughput streams. Setting a specific GRACE PERIOD is recommended to reduce high disk usage.
If you specify a GRACE PERIOD for left/outer joins, the grace period defines when the left/outer join result is emitted. If you don't specify a GRACE PERIOD for left/outer joins, left/outer join results are emitted eagerly, which may cause "spurious" result records, so we recommended that you specify a GRACE PERIOD.
Accept events for up to two hours after the window ends. Events that arrive after the grace period has passed are dropped and not included in the aggregate result.
1 2 3 4