Apache Airflow 传感器

引言

Apache Airflow 是一个用于编排复杂工作流程的强大工具，而传感器是其最重要的组件之一。传感器充当 Airflow 中特殊的运算符，旨在等待特定事件或条件，然后才允许工作流程继续进行。无论是等待文件出现、监视资源可用性，还是等待外部事件完成，传感器对于确保工作流程仅在满足所有先决条件时执行至关重要。

什么是 Airflow 中的传感器？

传感器是 Airflow 运算符的一个独特类别。与执行任务并返回结果的标准运算符不同，传感器仅用于等待条件的满足。一旦条件满足，传感器即成功，允许其下游任务执行。

例如

• FileSensor 等待特定文件出现在指定位置。
• HttpSensor 检查 HTTP 端点是否可访问并返回预期响应。
• SqlSensor 确保满足数据库中的特定查询条件。

传感器的特点

• 专业用途： 传感器针对等待任务进行了优化，这使它们与标准运算符有所不同。
• 效率： 传感器通过在 poke 或 reschedule 模式下运行来最大限度地减少资源使用。
• 灵活性： 它们高度可配置，允许进行调整以满足特定的延迟和性能需求。
• 集成： 通过针对各种提供程序和系统的内置传感器，Airflow 支持广泛的用例。

传感器的操作模式

传感器以两种不同的模式运行：poke 和 reschedule。每种模式都有其用例和权衡，主要围绕资源利用率和延迟。

Poke 模式

• 定义： 在 poke 模式下，传感器在其整个运行时占用一个工作槽，不断检查是否满足条件。
• 用例： 适用于需要经常检查条件的情况，例如每秒检查一次，或者当低延迟至关重要时。
• 示例

Reschedule 模式

• 定义： 在 reschedule 模式下，传感器在检查之间放弃工作槽，在下一次检查之前休眠一段定义的时间。
• 特性
  • 低资源使用率（在休眠间隔期间释放工作程序）。
  • 较高的延迟（较不频繁的检查可能会延迟满足条件的检测）。
• 用例： 适用于对时间要求不高的条件，例如每分钟检查一次，或者当资源效率是优先事项时。
• 示例

配置参数

配置传感器时，通常使用以下参数

• poke_interval： 检查之间的间隔（秒）。
• 超时： 传感器在失败前将等待的最长时间（秒）。
• 模式： 确定传感器是在 poke 还是 reschedule 模式下运行。
• task_id： DAG 中传感器的唯一标识符。

Airflow 中的传感器类型

Airflow 提供了各种预构建的传感器，涵盖了各种领域的常见用例。此外，可以为特定需求开发自定义传感器。以下是一些常用的传感器

文件传感器

• 目的： 监视本地或远程存储系统中文件的存在或修改。
• 示例
  • FileSensor：检查给定路径中是否存在文件。
  • HdfsSensor：监视 HDFS（Hadoop 分布式文件系统）中的文件。

数据库传感器

• 目的： 等待数据库中的特定条件。
• 示例
  • SqlSensor：执行 SQL 查询并等待非空结果。
  • PostgresSensor：类似于 SqlSensor，但专为 PostgreSQL 数据库量身定制。

API 和 Web 传感器

• 目的： 监视 HTTP 端点或 API 的特定响应。
• 示例
  • HttpSensor：检查 HTTP 端点是否有特定响应。
  • WebhookSensor：等待传入的 webhook 事件。

云和外部服务传感器

• 目的： 与外部服务和云平台集成。
• 示例
  • S3PrefixSensor：等待 S3 存储桶中具有特定前缀的密钥。
  • GcsObjectSensor：检查 Google Cloud Storage 中是否存在文件。
  • PubSubPullSensor：等待 Google Cloud Pub/Sub 主题上的消息。

基于时间的传感器

• 目的： 确保工作流程在特定时间或延迟后执行。
• 示例
  • TimeDeltaSensor：等待定义的时间间隔。
  • TimeSensor：仅在一天中的指定时间之后触发任务。

使用传感器的最佳实践

有效使用传感器可确保最佳的资源利用率和可靠的工作流程执行。以下是一些最佳实践

选择正确的模式

• 对于低延迟、高频率的检查，使用 poke 模式。
• 对于不太频繁的检查，选择 reschedule 模式以减少工作槽的使用。

设置适当的超时

• 始终定义合理的超时时间，以防止任务无限期挂起。

优化 poke_interval

• 根据条件的敏感度调整 poke_interval。
• 对于高优先级任务，使用较短的间隔；对于非关键任务，使用较长的间隔。

利用内置传感器

• 使用 Airflow 丰富的预构建传感器库，以节省开发时间并减少错误。
• 确保与您的系统兼容并适当地配置传感器。

避免过度使用传感器

• 限制 DAG 中传感器的数量，以减少资源争用。
• 如果可能，将相关条件合并到单个传感器中。

如有必要，实施自定义传感器

• 对于特定用例，扩展 BaseSensorOperator 类以创建自定义传感器。

常见陷阱及如何避免

1. 过度使用资源
   • 过度使用 poke 模式下的传感器可能会耗尽工作槽。如果可能，请使用 reschedule 模式。
2. 低效的轮询
   • 为非关键任务设置过短的 poke_interval 值会增加资源消耗。使用适当的间隔。
3. 忽略超时
   • 未能设置超时可能会导致任务无限期挂起，从而导致下游延迟。
4. 冗余检查
   • 避免多个传感器检查相同条件。尽可能合并逻辑。
   • Apache Airflow 传感器对于构建强大而高效的工作流程至关重要。
   • 通过在继续之前等待特定条件，传感器可确保工作流程以正确的顺序并在正确的条件下执行。

了解 poke 和 reschedule 模式之间的区别，选择适当的内置传感器，并遵循最佳实践，可以显着提高工作流程的性能和可靠性。