如何在 Airflow 中实现轮询？
How to implement polling in Airflow?
我想用 Airflow 实现定期轮询外部系统（ftp 服务器等）的数据流，检查符合特定条件的新文件，然后 运行 一堆任务那些文件。现在，我是 Airflow 的新手，读到传感器是你会在这种情况下使用的东西，我实际上设法编写了一个传感器，当我 运行 "airflow test" 时它工作正常.但是我对传感器和 DAG 调度的 poke_interval 的关系有点困惑。我应该如何为我的用例定义这些设置？或者我应该使用其他方法吗？我只希望 Airflow 在这些文件可用时 运行 执行任务，而不是在一段时间内没有新文件可用时让仪表板充满故障。
您的理解是正确的，当您想要进行轮询时，使用传感器是可行的方法，既可以使用现有传感器，也可以实现您自己的传感器。
但是，它们始终是 DAG 的一部分，并且不会在其边界之外执行。 DAG 执行取决于 start_date 和 schedule_interval，但您可以利用它和传感器根据外部服务器的状态来实现某种 DAG：一种可能的方法是启动整个 DAG一个传感器，它检查条件是否发生，如果条件不满足则决定跳过整个 DAG（您可以确保传感器将下游任务标记为 skipped 而不是 failed 通过设置它们的 soft_fail 参数到 True)。通过使用最频繁的计划选项 (* * * * *)，您可以设置一分钟的轮询间隔。如果您真的需要最短的轮询时间，您可以调整传感器的poke_interval和timeout参数。
但是请记住，Airflow 本身可能无法保证执行时间，因此对于非常短的轮询时间，您可能需要研究替代方案（或者至少考虑与我刚刚分享的方法不同的方法） .
我想用 Airflow 实现定期轮询外部系统（ftp 服务器等）的数据流，检查符合特定条件的新文件，然后 运行 一堆任务那些文件。现在，我是 Airflow 的新手，读到传感器是你会在这种情况下使用的东西，我实际上设法编写了一个传感器，当我 运行 "airflow test" 时它工作正常.但是我对传感器和 DAG 调度的 poke_interval 的关系有点困惑。我应该如何为我的用例定义这些设置？或者我应该使用其他方法吗？我只希望 Airflow 在这些文件可用时 运行 执行任务，而不是在一段时间内没有新文件可用时让仪表板充满故障。
您的理解是正确的，当您想要进行轮询时，使用传感器是可行的方法，既可以使用现有传感器，也可以实现您自己的传感器。
但是，它们始终是 DAG 的一部分，并且不会在其边界之外执行。 DAG 执行取决于 start_date 和 schedule_interval，但您可以利用它和传感器根据外部服务器的状态来实现某种 DAG：一种可能的方法是启动整个 DAG一个传感器，它检查条件是否发生，如果条件不满足则决定跳过整个 DAG（您可以确保传感器将下游任务标记为 skipped 而不是 failed 通过设置它们的 soft_fail 参数到 True)。通过使用最频繁的计划选项 (* * * * *)，您可以设置一分钟的轮询间隔。如果您真的需要最短的轮询时间，您可以调整传感器的poke_interval和timeout参数。
但是请记住，Airflow 本身可能无法保证执行时间，因此对于非常短的轮询时间，您可能需要研究替代方案（或者至少考虑与我刚刚分享的方法不同的方法） .