如何在Airflow中结合虚拟Python环境使用@task.sensor装饰器

要让Python传感器在虚拟环境中运行（依赖Airflow主环境未安装的包，比如pymsteams），可以通过以下两种方式实现：

方法一：用@task.virtualenv直接定义传感器 #

@task.virtualenv装饰器支持sensor=True参数，开启后该任务会以传感器模式运行，同时自动创建指定依赖的虚拟环境。

修改你的传感器函数如下：

@task.virtualenv(requirements=["pymsteams"], sensor=True, poke_interval=60)
def virtual_env_sensor():
    import pymsteams
    # 这里编写传感器的轮询逻辑，返回True表示满足触发条件，False则继续等待
    return True

• sensor=True：标记该任务为传感器类型
• poke_interval：设置传感器的轮询间隔（单位：秒），可根据需求调整
• 其他@task.virtualenv的参数（如python_version）也可按需添加

方法二：给@task.sensor配置虚拟环境执行器 #

如果你偏好使用@task.sensor装饰器，可以通过executor_config指定虚拟环境的配置：

@task.sensor(
    poke_interval=60,
    executor_config={
        "virtualenv": {
            "requirements": ["pymsteams"],
            # 可选：指定虚拟环境使用的Python版本
            # "python_version": "3.9"
        }
    }
)
def virtual_env_sensor():
    import pymsteams
    return True

修改后的完整DAG示例 #

import datetime
from airflow.decorators import dag, task


@task.virtualenv(requirements=["pymsteams"])
def virtual_env_task():
    import pymsteams
    # 任务逻辑
    print("虚拟环境任务执行完成")


@task.virtualenv(requirements=["pymsteams"], sensor=True, poke_interval=60)
def virtual_env_sensor():
    import pymsteams
    # 传感器逻辑示例：这里直接返回True表示满足条件
    return True


@dag(start_date=datetime.datetime(2024, 1, 1), catchup=False)
def virtual_env_task_dag():
    virtual_env_task()


@dag(start_date=datetime.datetime(2024, 1, 1), catchup=False)
def virtual_env_sensor_and_task_dag():
    virtual_env_sensor() >> virtual_env_task()


virtual_env_task_dag()
virtual_env_sensor_and_task_dag()

修改后，virtual_env_sensor_and_task_dag会在虚拟环境中运行传感器和任务，不会再出现pymsteams导入失败的问题。

内容的提问来源于stack exchange，提问作者user3124206