KSQLDB Select Table Query and Persist Results in Kafka

ksqlDB
1

Hello,

We have a KsqlDB Table select query that we want to use in our pipeline. But the result of our select query isn’t persisted in kafka topics, so we can’t continue to the flow. We can’t find any solution in the docs and getting this error if we try to create Stream from Table:

error message : Invalid result type. Your SELECT query produces a TABLE. Please use CREATE TABLE AS SELECT statement instead

OUR TABLE CREATE QUERY (works fine, no errors occured here, we can select expected results in cli):

CREATE TABLE ISACTIVE_TABLE_TEST3 WITH (
  FORMAT = 'AVRO'
  ) AS
SELECT AFTER->Id, LATEST_BY_OFFSET(AFTER->IsActive, 2)[1] AS Before_IsActive,
  LATEST_BY_OFFSET(AFTER->IsActive, 2)[2] AS After_IsActive
FROM test_table_stream2
GROUP BY AFTER->Id;

OUR CREATE STREAM QUERY WHICH WE TRY TO PERSIST THE SELECT QUERY RESULT

CREATE STREAM IsActiveChange_Test AS
SELECT Id, AFTER_ISACTIVE FROM ISACTIVE_TABLE_TEST3 
WHERE AFTER_ISACTIVE IS NOT null AND BEFORE_ISACTIVE != AFTER_ISACTIVE 
EMIT CHANGES;

Is table’s usage is restricted with stream joins and CLI selects or could we create a stream with querying tables?
Thanks in advance.

Batu

2

In your second query, the input is TABLE ISACTIVE_TABLE_TEST3 and you only apply a filter on the table. Thus, the result is a TABLE, and you would need to use CREATE TABLE ... AS SELECT ....

The changelog stream of the created table will be written into a corresponding Kafka topic for this case.

But the result of our select query isn’t persisted in kafka topics, so we can’t continue to the flow.

Not sure what you mean by this? You can always deploy a persistent query via CREATE STREAM/TABLE ... AS SELECT to write the result into a Kafka topic.

3

This topic was automatically closed after 30 days. New replies are no longer allowed.