如何在Airflow的PythonVirtualenvOperator中获取执行日期上下文？

核心实现逻辑 #

• Airflow 2.x 中provide_context参数已废弃，PythonVirtualenvOperator作为隔离运行的任务，无法直接接收外部的Airflow上下文对象，需要通过模板参数传递执行日期字符串，在虚拟环境内用pendulum解析后使用即可。
• 直接使用Airflow内置模板变量{{ execution_date }}（2.3+版本也可以用等价的{{ logical_date }}），该变量会在任务运行时自动解析为当前任务的执行时间字符串，通过op_kwargs传入虚拟环境。

完整可运行示例 #

from airflow import DAG
from airflow.operators.python import PythonVirtualenvOperator
import pendulum

# 虚拟环境内运行的函数
def exec_in_venv(exec_date_str):
    # 虚拟环境内导入对应版本的pendulum
    import pendulum
    # 解析字符串为pendulum datetime对象
    exec_date = pendulum.parse(exec_date_str)
    # 后续业务逻辑可直接使用该对象
    print(f"任务执行日期：{exec_date.to_date_string()}")
    print(f"带时区完整时间：{exec_date.to_datetime_string()}")
    return exec_date.int_timestamp

with DAG(
    dag_id="venv_get_exec_date_demo",
    start_date=pendulum.datetime(2023, 1, 1, tz="Asia/Shanghai"),
    schedule_interval="0 2 * * *",
    catchup=False
) as dag:
    test_venv_task = PythonVirtualenvOperator(
        task_id="get_exec_date_test",
        python_callable=exec_in_venv,
        op_kwargs={
            # 传入模板化的执行日期字符串
            "exec_date_str": "{{ execution_date }}"
        },
        # 声明虚拟环境需要安装的依赖，包含pendulum
        requirements=[
            "pendulum>=2.1.2"
        ],
        python_version="3.9"
    )

注意事项 #

• 不要尝试直接传递Airflow的上下文对象进入虚拟环境，隔离环境无法识别外部的Airflow自定义类，传递字符串是最稳定的方案。
• 虚拟环境内的pendulum版本不需要和Airflow自带的pendulum版本一致，字符串解析的兼容性已经覆盖绝大多数常用时间格式，不需要额外做格式转换。

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