我正试图创建一个ksql表来保存实体的最新版本。我想知道推荐的方法是什么。
到目前为止，我已经尝试了以下操作：我通过debezium将发票加载到kafka中，然后我创建了以下流，以便能够从ksql:使用它们
ksql> create stream invoice_stream with (kafka_topic='dbserver1.invoices.invoice', value_format='AVRO');
Debezium添加了关于数据库表行前后状态的数据，这让它有点难以使用，所以我在上面创建了另一个流，只获取我感兴趣的数据：
create stream invoice
with (kafka_topic='invoice', value_format='AVRO')
as
select i.before->id as before_id,
i.after->id as after_id,
ifnull(i.transaction->id, 'NA') as transaction_id,
i.after->description as description,
i.after->invoice_date as invoice_date,
i.after->status as status
from invoice_stream i;
到目前为止，一切都很好，我可以通过推送查询来查询流，并查看预期结果：
ksql> select * from invoice emit changes;
+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+
|ROWTIME |ROWKEY |BEFORE_ID |AFTER_ID |TRANSACTION_ID |DESCRIPTION |INVOICE_DATE |STATUS |
+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+
|1583961059498 | |null |1 |NA |Invoice A |18201 |N |
|1583961059499 | |null |2 |NA |Invoice B |18205 |N |
|1583961059499 | |null |3 |NA |Invoice C |18210 |N |
|1583961059499 | |null |4 |NA |Invoice D |18215 |N |
|1583961263233 | |null |5 |623 |test line added later |18263 |N |
|1584007291546 | |5 |5 |625 |test line added later |18263 |P |
由于没有密钥，我在指定分区的顶部创建了另一个流：
ksql> create stream invoice_rekeyed as select * from invoice partition by after_id;
ksql> describe invoice_rekeyed;
Name : INVOICE_REKEYED
Field | Type
--------------------------------------------
ROWTIME | BIGINT (system)
ROWKEY | VARCHAR(STRING) (system)
BEFORE_ID | INTEGER
AFTER_ID | INTEGER (key)
TRANSACTION_ID | VARCHAR(STRING)
DESCRIPTION | VARCHAR(STRING)
INVOICE_DATE | INTEGER
STATUS | VARCHAR(STRING)
--------------------------------------------
最后，我创建了一个这样的表：
create table invoice_table(before_id int, after_id int, transaction_id string, description string, invoice_date int, status string)
with (kafka_topic='INVOICE_REKEYED', key='after_id', value_format='AVRO');
因此，在这一点上，我希望能够通过rowkey查询表，但我得到了以下消息：
ksql> select * from invoice_table where rowkey = 5;
Table 'INVOICE_TABLE' is not materialized. Refer to https://cnfl.io/queries for info on query types. If you intended to issue a push query, resubmit with the EMIT CHANGES clause
KSQL currently only supports pull queries on materialized aggregate tables. i.e. those created by a 'CREATE TABLE AS SELECT <fields>, <aggregate_functions> FROM <sources> GROUP BY <key>' style statement.
Query syntax in KSQL has changed. There are now two broad categories of queries:
- Pull queries: query the current state of the system, return a result, and terminate.
- Push queries: query the state of the system in motion and continue to output results until they meet a LIMIT condition or are terminated by the user.
'EMIT CHANGES' is used to to indicate a query is a push query. To convert a pull query into a push query, which was the default behavior in older versions of KSQL, add `EMIT CHANGES` to the end of the statement
before any LIMIT clause.
For example, the following are pull queries:
'SELECT * FROM X WHERE ROWKEY=Y;' (non-windowed table)
'SELECT * FROM X WHERE ROWKEY=Y AND WINDOWSTART>=Z;' (windowed table)
The following is a push query:
'SELECT * FROM X EMIT CHANGES;'
Note: Persistent queries, e.g. `CREATE TABLE AS ...`, have an implicit `EMIT CHANGES`, but we recommend adding `EMIT CHANGES` to these statements.
此外，如果我将其作为推送查询进行查询，我会看到关键字5:的不止一行
ksql> select * from invoice_table emit changes;
+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+
|ROWTIME |ROWKEY |BEFORE_ID |AFTER_ID |TRANSACTION_ID |DESCRIPTION |INVOICE_DATE |STATUS |
+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+------------------------+
|1583961059498 |1 |null |1 |NA |Invoice A |18201 |N |
|1583961059499 |2 |null |2 |NA |Invoice B |18205 |N |
|1583961059499 |3 |null |3 |NA |Invoice C |18210 |N |
|1583961059499 |4 |null |4 |NA |Invoice D |18215 |N |
|1583961263233 |5 |null |5 |623 |test line added later |18263 |N |
|1584007291546 |5 |5 |5 |625 |test line added later |18263 |P |
我想了解为什么表没有具体化，因为根据之前的消息，这似乎是阻止我开始按rowkey查询表的原因。
提前感谢
更新
尝试Robin的例子，我实际上得到了预期的行为；在该示例中，在更新原始数据库行的同时运行查询会显示以下更改：
ksql> select * from customers where id = 5 emit changes;
+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+
|ROWTIME |ROWKEY |ID |FIRST_NAME |LAST_NAME |EMAIL |GENDER |CLUB_STATUS |COMMENTS |CREATE_TS |UPDATE_TS |
+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+
|1584102664415 |5 |5 |Hansiain |Coda |hcoda4@senate.gov|Male |platinum |Centralized full-|2020-03-13T12:29:|2020-03-13T12:29:|
| | | | | | | | |range approach |53Z |53Z |
|1584102741712 |5 |5 |Rodrigo |Coda |hcoda4@senate.gov|Male |platinum |Centralized full-|2020-03-13T12:29:|2020-03-13T12:32:|
| | | | | | | | |range approach |53Z |21Z |
但是，如果查询被终止并再次运行，则只有最新版本可用：
ksql> select * from customers where id = 5 emit changes;
+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+
|ROWTIME |ROWKEY |ID |FIRST_NAME |LAST_NAME |EMAIL |GENDER |CLUB_STATUS |COMMENTS |CREATE_TS |UPDATE_TS |
+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+-----------------+
|1584102741712 |5 |5 |Rodrigo |Coda |hcoda4@senate.gov|Male |platinum |Centralized full-|2020-03-13T12:29:|2020-03-13T12:32:|
| | | | | | | | |range approach |53Z |21Z |
然而，原则上，在我的例子中做同样的事情，总是返回行的所有版本：
ksql> print 'dbserver1.invoices.invoice' from beginning limit 50;
Format:AVRO
3/13/20 12:23:09 PM UTC, 1, {"id": 1, "description": "Invoice A", "invoice_date": 18201, "status": "N", "__op": "r", "__ts_ms": 1584102188934, "__transaction_id": null}
3/13/20 12:23:09 PM UTC, 2, {"id": 2, "description": "Invoice B", "invoice_date": 18205, "status": "N", "__op": "r", "__ts_ms": 1584102188936, "__transaction_id": null}
3/13/20 12:23:09 PM UTC, 3, {"id": 3, "description": "Invoice C", "invoice_date": 18210, "status": "N", "__op": "r", "__ts_ms": 1584102188938, "__transaction_id": null}
3/13/20 12:23:09 PM UTC, 4, {"id": 4, "description": "Invoice D", "invoice_date": 18215, "status": "N", "__op": "r", "__ts_ms": 1584102188938, "__transaction_id": null}
^CTopic printing ceased
ksql> create table invoice_table with (kafka_topic='dbserver1.invoices.invoice', value_format='AVRO');
Message
---------------
Table created
---------------
ksql> select * from invoice_table where id = 4 emit changes;
+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+
|ROWTIME |ROWKEY |ID |DESCRIPTION |INVOICE_DATE |STATUS |__OP |__TS_MS |__TRANSACTION_ID |
+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+
|1584102189675 |4 |4 |Invoice D |18215 |N |r |1584102188938 |null |
|1584102365378 |4 |4 |Invoice D UPDATED |18215 |N |u |1584102365128 |623 |
^CQuery terminated
ksql> select * from invoice_table where id = 4 emit changes;
+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+
|ROWTIME |ROWKEY |ID |DESCRIPTION |INVOICE_DATE |STATUS |__OP |__TS_MS |__TRANSACTION_ID |
+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+---------------------+
|1584102189675 |4 |4 |Invoice D |18215 |N |r |1584102188938 |null |
|1584102365378 |4 |4 |Invoice D UPDATED |18215 |N |u |1584102365128 |623 |
有什么配置可以在行为上产生这种差异吗？
这里有一些东西需要取消拾取并帮助您，但最重要的答案是非物化表不支持拉式查询，而且您还没有物化它，在#3985交付之前也不能。
正如您所看到的，您可以对表运行推送查询。您看到的多个输出取决于状态的变化。如果您取消推送查询并重新运行它，您将只看到每个键的单个状态。
Debezium在使用之前和之后添加了有关数据库表行状态的数据
查看io.debezium.transforms.ExtractNewRecordState单一消息转换，它将使有效负载变平，并仅在消息中显示当前状态
'transforms'= 'unwrap',
'transforms.unwrap.type'= 'io.debezium.transforms.ExtractNewRecordState',
由于没有密钥，我在顶部创建了另一个流，我指定了分区：
这是一种方法，但更好的方法是将密钥设置为Kafka Connect摄取的一部分
'transforms'= 'extractkey',
'transforms.extractkey.type'= 'org.apache.kafka.connect.transforms.ExtractField$Key',
'transforms.extractkey.field'= 'id',
'key.converter'= 'org.apache.kafka.connect.storage.StringConverter',
有关这些活动的示例，请查看最近的QCon研讨会。