Airflow 传感器

Python 中的 Apache Airflow 入门

Mike Metzger

Data Engineer

传感器

  • 一种等待条件为真的算子
    • 文件被创建
    • 数据库记录被上传
    • Web 请求返回响应
  • 可设置检查条件为真的频率
  • 分配给任务使用

Airflow 传感器等待条件变为真的示意图

Python 中的 Apache Airflow 入门

传感器详解

  • 派生自 airflow.sdk.BaseSensorOperator
  • 传感器参数：
  • mode——检查条件的方式
    • mode='poke'——默认，反复运行
    • mode='reschedule'——释放任务槽，稍后重试
  • poke_interval——两次检查间隔
  • timeout——超时后任务失败
  • 还包含常规算子属性
Python 中的 Apache Airflow 入门

文件传感器

  • 属于 airflow.providers.standard.sensors 库
  • 检查指定位置是否存在文件
  • 也可检查目录中是否存在任意文件
from airflow.providers.standard.sensors.filesystem import FileSensor

file_sensor_task = FileSensor(task_id='file_sense',
                              filepath='salesdata.csv',
                              poke_interval=300,
                              timeout=3000
                              )

init_sales_cleanup() >> file_sensor_task >> generate_report()
Python 中的 Apache Airflow 入门

其他传感器

  • airflow.providers.*.sensors 中还有许多
  • ExternalTaskSensor——等待另一个 DAG 中的任务完成
  • HttpSensor——请求 URL 并检查内容
  • SqlSensor——运行 SQL 查询以检查内容
Python 中的 Apache Airflow 入门

何时使用传感器？

  • 何时条件为真不确定
  • 不希望立刻失败
  • 无需循环即可重复尝试

 

 

Airflow 传感器等待条件变为真的示意图

Python 中的 Apache Airflow 入门

开始练习！

Python 中的 Apache Airflow 入门

Preparing Video For Download...