我已经开始学习mqtt，因为我在当前的组织中有一个远程信息处理的用例。我想将mqtt代理（mosquitto）消息集成到我的kafka中。
由于每个车辆都在单个组织内的mqtt broker中以其自己的主题发送数据，因此我想将所有这些数据推送到kafka中。现在我知道在Kafka（超过一百万个）里创造这么多主题是不明智的。另外，我不想把所有的车辆数据保存在一个Kafka主题中，因为我想稍后将所有这些数据放在s3中，通过车辆id进行区分。
我怎样才能做到这一点，而不是在Kafka这么多的主题。一种方法是Kafka的消费者隔离事件并放入s3，但我相信s3中会有很多小文件。
1条答案
按热度按时间ocebsuys1#
通常，如果您有相同的逻辑实体，那么您将使用相同的主题。
您可以使用用于kafka connect的mqtt插件将数据从mqtt流式传输到kafka，以及kafka connect的单个消息转换
RegexRouter修改写入消息的主题名，其他smt修改消息键。这样，您就可以在一个主题中获得所有消息，并根据车辆id进行分区。这可能是存储它的最佳方式。从那里，你可以使用你想要的数据。当要将数据流传输到s3时，您可以使用kafka connect s3 sink，正如cricket\u007所提到的，如果您担心的只是数据量，那么可以按时间对数据进行分区。如果要将消息路由到不同的bucket或同一bucket的区域，可以使用流处理（例如kafka streams/ksqldb）来预处理主题以填充其他主题。
有关mqtt连接器的示例，请参见此处。