如何从与 KSQL 窗口表对应的 kafka 主题获取窗口开始或结束时间？答案

【问题标题】：How to get window start or end time from kafka topic corresponding to KSQL windowed table?如何从与 KSQL 窗口表对应的 kafka 主题获取窗口开始或结束时间？
【发布时间】：2020-05-14 00:23:42
【问题描述】：

我想从 kafka-console-consumer 命令获取窗口开始时间戳（这里是 1530008520000）。

它适用于 KSQL：

ksql> select * from DEV_MONITOR_RULE_2557_104782_233_2_TABLE;

1530008581051 | 2557 : Window{start=1530008520000 end=-} | 2557 | 2 | 1530008581051

但它不适用于 kafka-console-consumer？

./bin/kafka-console-consumer --zookeeper 10.12.0.157:2181 --topic DEV_MONITOR_RULE_2557_104782_233_2_TABLE

{"HITCOUNTS":2,"TENANTID":2557,"HITTIME":1530008581051}

如何从 kafka-console-consumer 打印窗口开始时间（这里是 1530008520000）？

谢谢！

【问题讨论】：

抱歉，您的问题是什么？您的 SELECT 查询已包含此信息，即 1530008581051?
如何从 kafka-console-consumer 打印窗口开始时间（这里是 1530008520000）而不是 HITTIME 1530008581051？

标签： apache-kafka ksqldb

【解决方案1】：

窗口开始时间体现在KSQL消息的ROWTIME，以及Kafka消息的时间戳中。您可以使用标准 API 访问此时间戳以处理 Kafka 消息。

AFAIK kafka-console-consumer 不支持显示时间戳。但是像kafkacat 这样的东西可以。

这是 KSQL 中的一些聚合数据，ROWTIME 显示窗口的开头（TIMESTAMPTOSTRING 用于漂亮地打印它）：

ksql> SELECT TIMESTAMPTOSTRING(ROWTIME, 'yyyy-MM-dd HH:mm:ss'), ROWTIME, STARS, STAR_COUNT FROM RATINGS_AGG2;
2018-06-27 09:30:00 | 1530091800000 | 1 | 2
2018-06-27 09:30:00 | 1530091800000 | 4 | 6
2018-06-27 09:30:00 | 1530091800000 | 2 | 2
2018-06-27 09:30:00 | 1530091800000 | 3 | 3

现在kafkacat同题：

$ kafkacat -b localhost:9092 -C -K: \
  -f '\nTimestamp: %T\t\tValue (%S bytes): %s' \
  -t RATINGS_AGG2

Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":1,"STARS":1}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":2,"STARS":1}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":3,"STARS":3}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":5,"STARS":1}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":6,"STARS":3}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":7,"STARS":1}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":7,"STARS":3}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":9,"STARS":1}
Timestamp: 1530091800000                Value (26 bytes): {"STAR_COUNT":9,"STARS":3}

【讨论】：

【解决方案2】：

您是否尝试过 kafka-console-consumer 选项print.timestamp=true？

$ kafka-console-consumer --property print.timestamp=true ...

【讨论】：