我知道这是一个非常不同的elasticsearch用例，我需要你的帮助。
主体结构（不能改变）：
有一些物理机器，我们有传感器。来自这些传感器的数据将传送到aws greengrass。
然后，使用lambda函数数据将通过使用mqtt进行elasticsearch。elasticsearch正在docker上运行。
这就是结构，在这里之前一切都准备好了✅
现在，在es的顶部，我需要一些软件，可以使用mqtt将这些数据发送到云数据库，例如dynamodb。
但这不是一次性迁移。它应该连续发送数据。基本上，我需要一个es和aws dynamodb之间的通道。
而且，传感器产生了如此多的数据，我们不想把它们全部存储在云中，但我们想把它们存储在es中。在将数据发送到云之前，需要在elasticsearch端进行一些过滤。就像“将每10个数据保存到云”那样，我们只能保存10个数据中的1个。
你知道怎么做吗？我没有这方面的经验，这看起来是一项具有挑战性的任务。我很想从这些领域有经验的人那里得到一些建议。
多谢？？