能否将Airflow模板变量以实际类型传入Operator？

问题场景 #

你希望将Airflow的模板变量（比如ts）以原生datetime类型传给PythonOperator的调用函数，而非字符串类型。直接传入{{ ts }}会触发变量未定义错误，包裹成字符串"{{ ts }}"传递后，函数又会因接收的是字符串而非datetime对象报错，且不想在函数里额外做类型转换。

解决方案 #

方案1：直接使用任务上下文的execution_date（推荐）

Airflow会自动将任务上下文（包含执行时间等元数据）传递给PythonOperator的调用函数，你可以修改函数接收**kwargs参数，直接获取原生的datetime对象：

def get_day(**kwargs):
    # execution_date 是原生datetime对象，可直接调用day属性
    return kwargs['execution_date'].day

py_op = PythonOperator(
    task_id='foo',
    python_callable=get_day,
    # Airflow 2.x及以上版本默认开启provide_context，可省略该参数
    dag=dag
)

方案2：用模板过滤器将ts转为datetime

如果必须使用ts模板变量，可以通过Airflow内置的to_datetime过滤器，在模板渲染时直接将字符串转为datetime对象：

def get_day(datetime_object):
    return datetime_object.day

py_op = PythonOperator(
    task_id='foo',
    python_callable=get_day,
    op_args=[
        "{{ ts | to_datetime }}"
    ],
    dag=dag
)

方案3：通过op_kwargs传递模板化的datetime对象

也可以用op_kwargs参数传递，同样借助to_datetime过滤器：

def get_day(datetime_object):
    return datetime_object.day

py_op = PythonOperator(
    task_id='foo',
    python_callable=get_day,
    op_kwargs={
        "datetime_object": "{{ ts | to_datetime }}"
    },
    dag=dag
)

补充说明 #

• 裸写{{ ts }}报错的原因：Python解释器会把它当成本地变量解析，而此时该变量未定义；只有将模板语法放在字符串中，Airflow的模板引擎才会正确渲染它。
• 包裹成字符串后，ts会被渲染为YYYY-MM-DDTHH:MM:SS格式的字符串，所以函数接收到的是字符串类型，自然没有day属性。
• 方案1是最优解，因为Airflow本身就提供了原生的datetime类型执行时间对象，无需额外转换操作。

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