How to update a running persistent query¶
You have a persistent query that is running and processing rows, and you want to change it to do something different. You want the new version of the query to start processing rows where the old version left off. This is generally known as upgrading a query.
1 2 3 4 5 6 7 8 9 10 11
ksqlDB provides two mechanisms to change a query that is already running:
- In-place upgrades: you modify the behavior of a query, and it resumes from a
previously committed offset. The syntax that ksqlDB uses to indicate an
in-place upgrade is
CREATE OR REPLACE.
- Replacing upgrades: you tear down an existing query,
and start a new one from either
latestoffsets. To accomplish this, you first issue a
TERMINATE <query_id>;and a
DROP <source>before creating the query again.
Obviously, it would be preferable to always perform an in-place upgrade when you change a query. But because of how streaming programs are constructed, this is not always possible.
To better understand the different types of upgrades that are allowed on persistent queries, here's a taxonomy using the combination of three types of query characteristics: source query, upgrade and (optionally) environment.
|Stateful queries maintain local storage
|Windowed queries maintain a limited amount of state specified by a window in time
|Joined queries read from multiple sources
|Multistage queries contain intermediate, non-user visible topics in Kafka
|Nondeterministic queries may produce different results when executing identical input
|Queries with none of the above characteristics
|Transparent upgrades change the way something is computed (e.g. improving a UDF performance)
|Data selecting query upgrades change which/how many events are emitted
|Schema evolving query upgrades change the output type of the data
|These upgrades change the source data, whether by means of modifying a JOIN or swapping out a source
|These upgrades are invisible to the user, but change the topology, such as the number of sub-topologies or the ordering of operations (e.g. filter push down)
|Scaling upgrades change the physical properties of the query in order to enable better performance characteristics.
|Unsupported upgrades are ones that will semantically change the query in an unsupported way. There are no plans to implement these migrations.
|Backfill requires the output data to be accurate not just from a point in time, but from the earliest point of retained history
|Cascading environments contain queries that are not terminal, but rather feed into downstream stream processing tasks
|Exactly Once environments do not allow for data duplication or missed events
|Ordered environments require that a single offset delineates pre- and post-migration (no events are interleaved)
|Live environments describe queries that cannot afford downtime, either by means of acting as live storage (e.g. responding to pull queries) or feeding into high availability systems (powering important functionality)
ksqlDB supports only in-place upgrades for data selection and schema evolution upgrades on a limited subset of query characteristics. ksqlDB doesn't guarantee validity of any environments when performing an in-place upgrade.
Any in place upgrades on windowed or joined sources, as well as upgrades on any table aggregation, are not yet supported.
Imagine a query that reads from a stream of purchases made at ksqlDB's flagship store, ksqlMart, and filters out transactions that might be invalid:
1 2 3 4 5 6 7 8 9 10 11
Over time, ksqlMart changes its return policy and begins issuing full refunds.
These records have a negative
cost column value. Since these records are now
valid, ksqlMart needs to update the query to remove the
cost > 0.00 clause:
1 2 3 4
CREATE OR REPLACE statement instructs ksqlDB to terminate the old query,
and create a new one that will continue from the last
record that the previous query processed. Note that this means any previously
processed data with negative cost will not be included, even if issuing the
This query upgrade is a simple, data selecting upgrade because it doesn't involve any aggregations; the only change is the criteria to emit rows. ksqlDB supports nearly all data selection modifications on source queries.
Over time, ksqlMart gets more sophisticated in their usage of Kafka to
monitor their input. They start publishing a new field to the
popularity. In order to reflect this change in their
stream, they need to issue two different commands:
1 2 3 4 5 6 7 8 9 10 11 12
There are a few things to note in the above statements:
- Data declaration statements can be updated using
CREATE OR REPLACE.
- ksqlMart re-issued the
SELECT *statement even though the statement text is identical to the previous statement they issued. This is necessary because ksqlDB resolves
SELECT *at the time the query was issued, which means that any updates to
purchasesafter issuing a
CREATE AS SELECTstatement aren't picked up in
Schema Evolution upgrades have much stricter requirements than Data Selection upgrades. ksqlDB supports only adding new columns at the end of the schema. Removing, renaming, or changing the type of any existing field is invalid.
Stateful data selection¶
The previous examples all involve stateless upgrades, but ksqlDB also enables data selection and limited schema evolution upgrades on some stateful queries. ksqlMart, as is common with data-driven companies that leverage ksqlDB, also has queries that generate analytics on their purchases:
1 2 3 4 5 6
After some time, they realize that the
purchase_stats stream doesn't account
properly for refunds. They're OK with having the initial purchase count toward
purchase_stats, but they don't want the refund to increment the
aggregation, so they update their query in place to add a filter for this
1 2 3 4 5 6 7
This updated query ensures only that new refunds don't count toward the stats, but anything that was counted before will remain.
If ksqlMart wanted to backfill the data properly, they would need to issue a
replacing upgrade that read from the earliest offset in the
1 2 3 4 5 6 7 8 9 10 11 12 13 14
This solution becomes more difficult if there are downstream consumers of the
Additional restrictions on stateful data selection¶
ksqlDB maintains state in order to accomplish stateful tasks such as aggregations. To ensure that all intermediate state is compatible, ksqlDB ensures that the intermediate schema is identical when changing a filter, which means that ksqlMart can only change filters to include fields that are already selected and can't remove a filter which is the only reference to a field.