环境变量名称不会被气流 DAG 中的值替换 [重复]

Posted 2023-03-25

【中文标题】环境变量名称不会被气流 DAG 中的值替换 [重复]

【英文标题】：Environment variables names are not replaced with values in airflow DAG [duplicate]

【发布时间】：2019-10-31 13:10:19

【问题描述】：

我正在尝试在气流 dag 中执行 SQL 代码。我能做到的。 我必须将环境变量名称放在我的 SQL 代码中，并且在执行 DAG 期间，变量的名称应该替换为变量的值。

tryMyDag.py

projectId = "`" + os.environ["GCP_PROJECT_ENV"]

stockLevels = BigQueryOperator(
    task_id='trafficSignals',
      use_legacy_sql=False,
      sql='/dags/sqls/trafficSignals_v1.sql',

trafficSignals.sql

select * from  projectId.dataset_name.trafficSignals_vw`

【问题讨论】：

感谢您的建议。

【参考方案1】：

您正在尝试在 sql 文件中使用环境变量？如果是这样，我认为这是不可能的，但是您可以使用带有 SQL 代码的 Python 文件，只需将您的 dags/sqls/trafficSignals_v1.sql 替换为 dags/sqls/traffic_signals_v1.py，并在其中放入一个变量：

select_trafficSignals = """
select * from  projectId.dataset_name.trafficSignals_vw
...
place you need a os env
""".format(os.environ['...'])

然后在tryMyDag.py引用这个文本变量：

from sqls.traffic_signals_v1 import select_trafficSignals

stockLevels = BigQueryOperator(
    task_id='trafficSignals',
      use_legacy_sql=False,
      sql=select_trafficSignals,

【参考方案2】：

感谢雷纳托。我使用了用户定义的宏并解决了这个问题。但是有多种方法可以解决这个问题。我找到了。气流变量可以在 .py 中定义，并在 .sql 中作为var.value.projectId 使用。感谢您的回复。