DAG 文件中的气流变量用法

Question

我有一个简短的 DAG，我需要在其中获取存储在 Airflow 中的变量（Airflow -> Admin -> 变量）。但是，当我们用作模板时，我遇到了错误。 示例代码和错误如下所示：

from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator

def display_variable():
    my_var = Variable.get("my_var")
    print('variable' + my_var)
    return my_var

def display_variable1():
    my_var = {{ var.value.my_var }}
    print('variable' + my_var)
    return my_var

dag = DAG(dag_id="variable_dag", start_date=airflow.utils.dates.days_ago(14),
    schedule_interval='@daily')

task = PythonOperator(task_id='display_variable', python_callable=display_variable, dag=dag)

task1 = PythonOperator(task_id='display_variable1', python_callable=display_variable1, dag=dag)

task >> task1

这里获取变量值的用法：

Variable.ger("my_var") --> is working

但是，我在使用其他方式时遇到了错误：

{{ var.value.my_var }}

错误：

  File "/home/airflow_home/dags/variable_dag.py", line 12, in display_variable1
    my_var = {{ var.value.my_var }}
    NameError: name 'var' is not defined

Answer 1

display_variable 两个函数都运行 Python 代码，因此 Variable.get() 可以按预期工作。 {{ ... }} 语法用于模板化字符串。大多数 Airflow 运算符的一些参数都支持模板化字符串，可以将其指定为 "{{ expression to be evaluated at runtime }}"。查找Jinja templating 了解更多信息。在执行任务之前，会评估模板化字符串。例如：

BashOperator(
    task_id="print_now",
    bash_command="echo It is currently {{ macros.datetime.now() }}",
)

Airflow 在执行之前评估bash_command，因此bash_command 将保持例如“今天是星期三”。

但是，像运行 Python 代码一样运行 {{ ... }} 实际上会尝试创建一个嵌套集：

{{ variable }}
^^
|└── inner set
|
outer set

由于 Python 中的集合不可散列，因此即使里面的语句有效，它也永远不会计算。

其他资源：

• https://www.astronomer.io/guides/templating
• 每个运算符上的template_fields 属性定义了哪些属性是模板可用的，请参阅您的运算符的文档以查看template_fields 的值：https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/operators/python/index.html#airflow.operators.python.PythonOperator.template_fields