我们如何记录将一组记录从 kafka 主题加载到数据库所花费的时间

Question

我正在使用 JDBC 接收器连接器将数据从 kafka 主题加载到 postgres。 在我的主题中，我有大约 10 万条记录。 现在需要知道我是否可以确定连接器何时开始加载记录并结束将所有这 10 万条记录加载到数据库中。

有可能吗？

Answer 1

所有消息都带有时间戳，因此您可以保留消息的时间戳。 添加到您的配置中

...
"transforms": "timestamp",
"transforms.timestamp.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.timestamp.target.type": "Timestamp",
"transforms.timestamp.field":"timestamp",
"transforms.timestamp.format": "yyyy-MM-dd HH:mm:ss"
...

您可能还需要将schema registry 配置为对应的value.converter。

但它会为您提供有关消息出现在 kafka 中的时间的信息。如果您想知道写入数据库的时间，那么只需创建一个默认值 = now() 的附加字段。然后您将获得数据库中数据记录的最新信息。