Airflow/Composer - 在 zip 打包的 DAG 中找不到模板答案

【问题标题】：Airflow/Composer - template not found in zip packaged DAGAirflow/Composer - 在 zip 打包的 DAG 中找不到模板
【发布时间】：2018-06-20 16:03:37
【问题描述】：

我无法让模板化的 SQL 文件在 Composer 中工作。我认为问题与我将 DAG 打包为 zip 文件以包含其他代码的事实有关。

我从这个开始（仅显示相关部分）：

dag = DAG('my_dag',
      default_args=default_args,
      schedule_interval=schedule_interval)

task0 = BigQueryOperator(
     task_id='task0',
     use_legacy_sql=False,
     bql='sql/query_file.sql',
     bigquery_conn_id=bigquery_conn_id,
     dag=dag)

文件结构如下：

/dags/my_dag_file.py
/dags/sql/query_file.sql
/dags/my_pkg/
/dags/my_pkg/__init__.py
/dags/my_pkg/extra_module.py

我正在像这样压缩它并将其复制到 Composer dags 文件夹中：

zip -r my_zip_file.zip *.py my_pkg/ sql/

这在本地工作，但在 Composer 上部署时出错：

TemplateNotFound: sql/query_file.sql

我确定我在 zip 中包含了 SQL 文件。我也尝试将它移动到根文件夹（没有 sql/ 子目录），但我得到了相同的结果。

我在某处读到在实例化 DAG 对象时需要设置 template_searchpath。我无法成功地做到这一点。当我尝试相对路径 (sql) 时，我会收到更多 TemplateNotFound 错误。当我尝试如下绝对路径时，我得到not a directory。

这是我尝试过的：

dag = DAG('my_dag',
      default_args=default_args,
      schedule_interval=schedule_interval,
      template_searchpath = os.path.dirname(__file__) + "/sql"
)

task0 = BigQueryOperator(
     task_id='task0',
     use_legacy_sql=False,
     bql='query_file.sql',
     bigquery_conn_id=bigquery_conn_id,
     dag=dag)

我还尝试将“sql”作为任务路径的一部分而不是模板搜索路径，然后我再次尝试将所有内容移至根级别，并得到相同的“不是目录”错误。

据我所知，问题与文件包含在 zip 中这一事实有关。 __file__ 返回/home/airflow/gcs/dags/my_zip_file.zip/my_dag_file.py。但随后os.listdir(os.path.dirname(__file__)) 抛出相同的not a directory 错误。所以也许是因为我们在一个 zip 存档中执行，我们不能以相同的方式使用文件夹和路径。也许Jinja被这件事绊倒了……？或者在打包 zip 文件时可能还有更多工作要做？

[2018-06-20 15:35:34,837] {base_task_runner.py:98} INFO - Subtask: Traceback (most recent call last):
[2018-06-20 15:35:34,838] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/bin/airflow", line 27, in <module>
[2018-06-20 15:35:34,840] {base_task_runner.py:98} INFO - Subtask:     args.func(args)
[2018-06-20 15:35:34,841] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/airflow/bin/cli.py", line 392, in run
[2018-06-20 15:35:34,841] {base_task_runner.py:98} INFO - Subtask:     pool=args.pool,
[2018-06-20 15:35:34,842] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/airflow/utils/db.py", line 50, in wrapper
[2018-06-20 15:35:34,843] {base_task_runner.py:98} INFO - Subtask:     result = func(*args, **kwargs)
[2018-06-20 15:35:34,843] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/airflow/models.py", line 1477, in _run_raw_task
[2018-06-20 15:35:34,844] {base_task_runner.py:98} INFO - Subtask:     self.render_templates()
[2018-06-20 15:35:34,844] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/airflow/models.py", line 1760, in render_templates
[2018-06-20 15:35:34,845] {base_task_runner.py:98} INFO - Subtask:     rendered_content = rt(attr, content, jinja_context)
[2018-06-20 15:35:34,847] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/airflow/models.py", line 2481, in render_template
[2018-06-20 15:35:34,848] {base_task_runner.py:98} INFO - Subtask:     return jinja_env.get_template(content).render(**context)
[2018-06-20 15:35:34,849] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/jinja2/environment.py", line 812, in get_template
[2018-06-20 15:35:34,849] {base_task_runner.py:98} INFO - Subtask:     return self._load_template(name, self.make_globals(globals))
[2018-06-20 15:35:34,850] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/jinja2/environment.py", line 774, in _load_template
[2018-06-20 15:35:34,851] {base_task_runner.py:98} INFO - Subtask:     cache_key = self.loader.get_source(self, name)[1]
[2018-06-20 15:35:34,852] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/jinja2/loaders.py", line 171, in get_source
[2018-06-20 15:35:34,854] {base_task_runner.py:98} INFO - Subtask:     f = open_if_exists(filename)
[2018-06-20 15:35:34,855] {base_task_runner.py:98} INFO - Subtask:   File "/usr/local/lib/python2.7/site-packages/jinja2/utils.py", line 151, in open_if_exists
[2018-06-20 15:35:34,856] {base_task_runner.py:98} INFO - Subtask:     return open(filename, mode)
[2018-06-20 15:35:34,856] {base_task_runner.py:98} INFO - Subtask: IOError: [Errno 20] Not a directory: '/home/airflow/gcs/dags/my_zip_file.zip/sql/query_file.sql'

【问题讨论】：

标签： python jinja2 airflow google-cloud-composer

【解决方案1】：

看起来 Airflow 当前（从 1.10 版本开始）不支持从压缩的 DAG 加载模板，因为它仅使用 jinja2.FileSystemLoader 来加载它们（请参阅 DAG#get_template_env）。

【讨论】：

【解决方案2】：

首先，确认 ZIP 中的文件结构符合预期。

那么，试试这个获取路径：

os.path.join(os.path.dirname(os.path.realpath(__file__)),"sql","query_file.sql")

这就是我们在 Airflow 部署中获取查询路径的方式。

【讨论】：

【解决方案3】：

使用文件的绝对路径更安全，如下所示 sql = os.path.abspath(os.path.join(os.path.dirname(__file__), "sql/query_file.sql"))

因为 Airflow 操作员/任务可能会在新创建的临时目录下运行您的命令/方法，该目录不会复制您的依赖项。查一下Implementation example on Airflow Github，你就明白了。

【讨论】：