如何为气流中的不同环境使用不同的配置文件？答案

【问题标题】：How to use different config files for different environments in airflow?如何为气流中的不同环境使用不同的配置文件？
【发布时间】：2022-06-19 14:24:28
【问题描述】：

我正在使用SparkKubernetesOperator，它有一个名为 application_file 的模板字段。通常在给这个字段一个文件名时，airflow 会读取该文件并模板化其中的 jinja 变量（就像 BashOperator 中的脚本字段一样）。

这样就可以了，文件信息显示在 Rendered Template 选项卡中，其中 jinja 变量替换为正确的值。

start_streaming = SparkKubernetesOperator(
    task_id='start_streaming',
    namespace='spark',
    application_file='user_profiles_streaming_dev.yaml',
    ...
    dag=dag,
)

我想在 application_file 字段中为不同的环境使用不同的文件

所以我在现场使用了 jinja 模板。但是当我用user_profiles_streaming_{{ var.value.env }}.yaml 更改application_file 时，呈现的输出只是user_profiles_streaming_dev.yaml 而不是文件内容。

我知道递归 jinja 变量替换在气流中是不可能的，但我想知道是否有任何解决方法可以使用不同的模板文件。

我尝试过的 - 我尝试使用不同的运算符并执行 xcom push 来读取文件内容并将其发送到 SparkKubernetesOperator。虽然这有利于根据环境读取不同的文件，但它并没有解决替换 jinja 变量的问题。

我还尝试制作一个自定义运算符，它继承了 SparkKubernetesOperator 并有一个 template_field applicaton_file_name 认为 jinja 替换将发生 2 次，但这也没有用。

【问题讨论】：

标签： jinja2 airflow

【解决方案1】：

我制作了一个包含环境详细信息 (dev/prod) 的 env 文件。然后我将此代码添加到我的 dag 文件的开头

ENV = None
with open('/home/airflow/env', 'r') as env_file:
    value = env_file.read()
    if value == None or value == "":
        raise Exception("ENV FILE NOT PRESENT")
    ENV = value

然后像这样访问代码中的环境

submit_job = SparkKubernetesOperator(
    task_id='submit_job',
    namespace="spark",
    application_file=f"adhoc_{ENV}.yaml",
    do_xcom_push=True,
    dag=dag,
)

这样我就可以拥有单独的 dev 和 prod 文件。

【讨论】：