我想对驻留在MongoDB中的数据执行一些转换，并通过Airflow DAG将转换后的数据插入ElasticSearch。 尽管我可以编写运算符来获取数据并执行转换，但挑战在于，我仅想在将新数据插入到集合中并获取该新数据以进行转换时才触发DAG。我已经阅读了有关Sensors的信息，但无法找到具体的解决方案。 有人可以请教吗？

编辑： 对于无法正确解释问题，我深表歉意。

以下是我要做的任务： 1.每当将新文档插入MongoDB时，都要获取文档并进行一些JSON转换。 2.将转换后的JSON数据流传输到完成进一步处理的下一个任务，然后将数据插入Elasticsearch索引。

我只提取新数据的想法是： 1.在MongoDB中插入唯一的增量batch_id以及数据。 2.获取上一个作业的最大batch_id，并为增加的batch_id戳MongoDB。 3.如果存在新的递增的batch_id，则获取数据。

以上想法可行，但我不确定这是否正确。

当前管道如下：