从 Kafka 获取最新价值

Question

我有一个名为 A 的 Kafka 主题。

主题 A 中的数据格式为：

{ id : 1, name:Whosebug, created_at:2017-09-28 22:30:00.000}
{ id : 2, name:confluent, created_at:2017-09-28 22:00:00.000}
{ id : 3, name:kafka, created_at:2017-09-28 24:42:00.000}
{ id : 4, name:apache, created_at:2017-09-28 24:41:00.000}

现在在消费者方面，我只想获取一小时的最新数据 window 意味着每隔一小时我需要从基于 created_at

的主题中获取最新值

我的预期输出是：

{ id : 1, name:Whosebug, created_at:2017-09-28 22:30:00.000}
{ id : 3, name:kafka, created_at:2017-09-28 24:42:00.000}

我认为这可以通过 ksql 解决，但我不确定。请帮助我。

提前致谢。

Answer 1

是的，您可以为此使用 KSQL。请尝试以下操作：

CREATE STREAM S1 (id BIGINT, name VARCHAR, created_at VARCHAT) WITH (kafka_topic = 'topic_name', value_format = 'JSON');

CREATE TABLE maxRow AS SELECT id, name, max(STRINGTOTIMESTAMP(created_at, 'yyyy-mm-dd hh:mm:ss.SSS')) AS creted_at FROM s1 WINDOW TUMBLING (size 1 hour) GROUP BY id, name;

结果将以 linux 时间戳格式显示 created_at 时间。您可以在新查询中使用 TIMESTAMPTOSTRING udf 将其更改为所需的格式。如果您发现任何问题，请告诉我。

从 Kafka 获取最新价值

Get Latest value from Kafka

apache-kafka

apache-kafka-streams

ksqldb