跳到主要内容

Airflow Sensor超时设置

在 Apache Airflow 中，Sensors 是一类特殊的任务，用于等待某些外部条件满足后再继续执行后续任务。例如，你可能需要等待某个文件出现在指定目录中，或者等待某个数据库表更新。然而，如果这些条件长时间未满足，任务可能会无限期等待，导致资源浪费。为了避免这种情况，Airflow 提供了超时设置功能。

什么是 Sensor 超时？​

Sensor 超时是指为 Sensor 任务设置一个最大等待时间。如果在超时时间内条件仍未满足，Sensor 任务将失败并停止等待。这有助于防止任务无限期挂起，确保工作流的健壮性。

如何设置 Sensor 超时？​

在 Airflow 中，可以通过 timeout 参数为 Sensor 设置超时时间。timeout 参数接受一个以秒为单位的整数值。以下是一个简单的示例：

python
from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
}

with DAG('file_sensor_example', default_args=default_args, schedule_interval=None) as dag:
    wait_for_file = FileSensor(
        task_id='wait_for_file',
        filepath='/path/to/your/file.txt',
        timeout=300,  # 设置超时时间为 300 秒（5 分钟）
        poke_interval=30,  # 每 30 秒检查一次文件是否存在
    )

在这个示例中，FileSensor 任务将等待 /path/to/your/file.txt 文件出现。如果在 300 秒（5 分钟）内文件未出现，任务将失败。

超时设置的实际应用场景​

假设你有一个工作流，需要等待某个外部系统生成一个报告文件，然后才能继续处理该文件。如果外部系统出现问题，文件可能永远不会生成。为了避免工作流无限期等待，你可以为 FileSensor 设置一个合理的超时时间。

python
from airflow import DAG
from airflow.sensors.filesystem import FileSensor
from airflow.utils.dates import days_ago

default_args = {
    'owner': 'airflow',
    'start_date': days_ago(1),
}

with DAG('report_processing', default_args=default_args, schedule_interval=None) as dag:
    wait_for_report = FileSensor(
        task_id='wait_for_report',
        filepath='/reports/daily_report.csv',
        timeout=1800,  # 设置超时时间为 1800 秒（30 分钟）
        poke_interval=60,  # 每 60 秒检查一次文件是否存在
    )

    # 后续任务
    process_report = BashOperator(
        task_id='process_report',
        bash_command='process_report.sh',
    )

    wait_for_report >> process_report

在这个案例中，wait_for_report 任务将等待 /reports/daily_report.csv 文件出现。如果在 30 分钟内文件未出现，任务将失败，并且不会执行后续的 process_report 任务。

超时设置的注意事项​

  • 合理设置超时时间：超时时间应根据实际业务需求设置。如果设置过短，可能会导致任务在条件即将满足时失败；如果设置过长，可能会导致资源浪费。
  • 结合 poke_interval 使用：poke_interval 参数控制 Sensor 检查条件的频率。合理设置 poke_interval 可以减少不必要的资源消耗。
  • 处理超时失败：当 Sensor 任务因超时而失败时，可以通过 Airflow 的重试机制或自定义逻辑来处理失败情况。

总结​

通过为 Airflow Sensors 设置超时时间，可以有效防止任务无限期等待，确保工作流的健壮性。合理设置 timeout 和 poke_interval 参数，可以优化资源使用并提高任务执行效率。

附加资源​

练习​

  1. 创建一个 DAG，使用 FileSensor 等待某个文件出现，并设置超时时间为 10 分钟。
  2. 修改上述 DAG，使 FileSensor 每 15 秒检查一次文件是否存在。
  3. 尝试为 FileSensor 设置不同的超时时间，观察任务的行为变化。