Airflow 不会让我将 try_number 宏转换为整数

如何解决Airflow 不会让我将 try_number 宏转换为整数

我的问题很简单，但没有真正提供真正的解决方案。我正在尝试构建的 DAG 将有一个关键的环境变量，该变量可能会根据它经历的重试次数而改变。例如，重试3次后，我们将切换下面代码中提到的环境变量"critical_value"值的值。

因此，当我推送下面的代码时，我收到以下损坏的 DAG 错误：

broken DAG: [/usr/local/airflow/dags/sync/dags/example_dag.py] invalid literal for int() with base 10: '{{ task_instance.try_number }}'

这是我们尝试将字符串转换为 int 时遇到的错误。在这种情况下，在运行时 '{{ task_instance.try_number }}' 将采用运行的尝试次数的适当值。我也尝试删除 int() 转换器，但随后在比较整数和字符串时出错。

任何帮助将不胜感激！提前致谢。

# required DAG arguments
default_args = {
    'owner':airflow,"depend_on_past":False,'start_date':datetime(2021,6,4),"retries": 3
}

# Instantiate DAG
with DAG(
    example_dag,default_args=default_args,schedule_interval=@once,) as dag:
    
    # Run python script of loader job
    full_script_task = KubernetesPodoperator(
        name=example_task,task_id=example_task_id,namespace=Variable.get("kubernetes_namespace"),image=some_image,configmaps=CONfig_MAPS,secrets=SECRETS,volumes=[logging_volume],volume_mounts=[logging_volume_mount],env_vars={
            "critical_value": "value_if_true" if int('{{ task_instance.try_number }}') >= 3 else "value_if_false"
        }
)

解决方法

Aiflow 可以处理 if 表达式并在 Jinja 中正确输入评估，但整个值需要在模板语法中。试试这个（根据 env_vars 的文档，list 需要是一个 KubernetesPodOperator，否则该值将是一个字符串化列表）：

env_vars=[
            {
                "critical_value": "{{ 'value_if_true' if task_instance.try_number >= 3 else 'value_if_false' }}"
            }
        ],

,

正如我们在您的测试中确认的那样，您不能在参数内的 if 子句中使用 jinja 模板化变量。你试过这个，变量被正确读取：

env_vars={
    "critical_value": "{{ task_instance.try_number }}"
}

您可以做的是处理运算符内部的 try_number，而不是对 critical_value 采取行动，只需使用 try_number。