Kafka连接MongoDB接收器连接器日期格式

Question

我使用的是Kafka connect Mongo DB接收器连接器(将Kafka到Mongo的消息持久化(。Kafka消息是简单的JSON:

{"field":"value", "ttl": "1648022542"}

正如您所看到的，我在每条消息中都有一个TTL字段。我想把这个TTL以日期格式保存在Mongo中(我希望这个文档在这个时候被删除，而Mongo只能用日期格式而不能用时间戳(。

有什么办法做到这一点吗？谢谢

Answer 1

我设法将时间戳的字符串表示转换为有效的ISODate(MongoDB可以在过滤器和查询中使用的有效时间戳(。

我使用的是Confluent Kafka v7.3.2和MongoDB Sink Connector，我在其中添加了上面提到的TimestampConverter属性。

这就是我的MongoSinkConnector属性文件的样子(它的相关部分(：

name=<mongo-sink-name>
topics=<topic-name>
connector.class=com.mongodb.kafka.connect.MongoSinkConnector
tasks.max=1
# Message types
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false
# Specific global MongoDB Sink Connector configuration
connection.uri=<mongodb-address>
database=<database-name>
collection=<collection-name>
# Single Message Transform (SMT) to convert "TimeStamp" string to ISODate
transforms=TimestampToIsoDate
transforms.TimestampToIsoDate.type=org.apache.kafka.connect.transforms.TimestampConverter$Value
transforms.TimestampToIsoDate.target.type=Timestamp
transforms.TimestampToIsoDate.field=TimeStamp
transforms.TimestampToIsoDate.format=yyyy-MM-dd'T'HH:mm:ss.SSSSS

因此，我认为在您的情况下，应该在转换中指定ttl。TimestampToIsoDate.field和转换中的一些数字表示。到IsoDate.format 的时间戳