如何在Airflow中实施轮询？

Question

我想使用Airflow实施数据流，这些数据流定期轮询外部系统（ftp服务器等），检查是否符合特定条件的新文件，然后为这些文件运行一堆任务。现在，我是Airflow的新手，并且读到Sensor是在这种情况下要使用的东西，实际上我设法编写了一个在运行“ airflow test”时可以正常工作的传感器。但是我对于传感器的poke_interval和DAG调度的关系有些困惑。如何为用例定义这些设置？还是应该使用其他方法？我只希望Airflow在这些文件可用时运行任务，而不是在一段时间内没有新文件可用时使仪表板充满故障。

Answer 1

您的理解是正确的，当您要轮询时，可以通过使用现有的传感器或实现自己的传感器来使用传感器。

但是，它们始终是DAG的一部分，并且不会在其边界之外执行。DAG的执行取决于start_date和schedule_interval，但是您可以利用此传感器和一个传感器来实现某种DAG，具体取决于外部服务器的状态：一种可能的方法是使用一个传感器来启动整个DAG，该传感器检查是否发生了状况，并且如果不满足该条件，则决定跳过整个DAG（您可以确保传感器将下游任务标记为，skipped而不是failed将其soft_fail参数设置为True）。通过使用最频繁的计划选项（* * * * *），您可以将轮询间隔设为一分钟。如果你真的需要一个最短轮询时间你可以调整传感器poke_interval和timeout参数。

但是请记住，Airflow本身并不能保证执行时间，因此对于非常短的轮询时间，您可能需要研究替代方法（或至少考虑与我刚刚分享的方法不同的方法）。