如何在 KSQL kafka 中获取所有历史记录

Question

我创建了一个流，我正在该流上编写 KSQL。 但是当我运行这个查询并且数据到达时，我可以看到记录，但是当我的数据没有到达并且我运行那个查询时，我看不到任何旧记录。

这就是我的 KSQL

LOG_DIR=./ksql_logs /usr/local/confluent/bin/ksql http://localhost:8088


CREATE STREAM AUDIT_EVENT ( ID VARCHAR , VERSION VARCHAR , ACTION_TYPE VARCHAR , EVENT_TYPE VARCHAR , CLIENT_ID VARCHAR , DETAILS VARCHAR , OBJECT_TYPE VARCHAR , UTC_DATE_TIME VARCHAR , POINT_IN_TIME_PRECISION VARCHAR , TIME_ZONE VARCHAR , TIMELINE_PRECISION VARCHAR , GROUP_ID VARCHAR , OBJECT_DISPLAY_NAME VARCHAR , OBJECT_ID VARCHAR , USER_DISPLAY_NAME VARCHAR , USER_ID VARCHAR , PARENT_EVENT_ID VARCHAR , NOTES VARCHAR , SUMMARY VARCHAR , AUDIT_EVENT_TO_UTC_DT VARCHAR , AUDIT_EVENT_TO_DATE_PITP VARCHAR , AUDIT_EVENT_TO_DATE_TZ VARCHAR , AUDIT_EVENT_TO_DATE_TP VARCHAR ) WITH (KAFKA_TOPIC='AVRO-AUDIT_EVENT', VALUE_FORMAT='AVRO');


SELECT * FROM "AUDIT_EVENT" WHERE CLIENT_ID='fgh-5d1e-17a2-9749-0e4d00';

我已经创建了表格并尝试过，但我的表格也看不到我的旧记录。

当我运行这个查询时，有什么方法可以记录吗？

Answer 1

SET 'auto.offset.reset' = 'earliest' 在您的 SELECT 查询语句之前。