MQTT 主题和 kafka 主题映射

Question

我已经开始了解 MQTT，因为我在当前组织中有一个远程信息处理用例。我想将 MQTT 代理（ mosquitto ）消息集成到我的 kafka 中。

由于每辆车都在单个组织内的 MQTT 代理中以自己的主题发送数据，因此我想将所有这些数据推送到 kafka 中。现在我知道在 kafka 中创建这么多主题（超过一百万）是不可取的。此外，我不想将所有车辆数据保存在一个 kafka 主题中，因为我想稍后将所有这些数据放在 S3 中，通过车辆 ID 进行区分。

如何在不使用 kafka 制作这么多主题的情况下实现这一目标。一种方法是 kafka 的消费者将事件隔离并放入 s3，但我相信 S3 中会有很多小文件。

Answer 1

通常，如果您有相同的逻辑实体，您将使用相同的主题。

您可以使用MQTT plugin for Kafka Connect 将数据从MQTT 流式传输到Kafka，并使用Kafka Connect 的Single Message Transform RegexRouter 来修改要写入消息的主题名称，并使用其他SMT 来修改消息密钥。这样您就可以在一个主题中获取所有消息，并根据车辆 ID 进行分区。这可能是最好的存储方式。

从那里，您可以随意使用数据。在将其流式传输到 S3 时，您可以使用Kafka Connect S3 sink，并且如 cricket_007 所述，如果您担心的只是音量，则按时间对数据进行分区。如果您想将消息路由到不同的存储桶或同一存储桶的区域，您可以使用流处理（例如 Kafka Streams / ksqlDB）来预处理主题以填充其他主题。

有关 MQTT 连接器的示例，请参阅 here。