У меня возникли проблемы с получением шаблонного файла 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:
zip -r my_zip_file.zip *.py my_pkg/ sql/
Это работало локально, но при развертывании в Composer было сделано сообщение об ошибке:
TemplateNotFound: sql/query_file.sql
Я уверен, что я включил файл SQL в zip. Я также попытался переместить его в корневую папку (без sql/subdirectory), но я получил тот же результат.
Я где-то читал, что вам нужно установить template_searchpath
при создании экземпляра объекта DAG. Я не смог сделать это успешно. Когда я пытаюсь относительный путь (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-архива, мы не можем использовать папки и пути одинаково. И, может быть, Джинджа справляется с этим...? Или, может быть, есть что-то еще делать при упаковке 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'
Похоже, что Airflow в настоящее время (начиная с версии 1.10) не поддерживает загрузку шаблонов из архивов DAG, поскольку для их загрузки используется только jinja2.FileSystemLoader
(см. DAG#get_template_env
).
Его безопаснее использовать абсолютный путь файла, например, ниже sql = os.path.abspath(os.path.join(os.path.dirname(__file__), "sql/query_file.sql"))
Поскольку оператор/задача Airflow, вероятно, запускает вашу команду/метод во вновь созданной временной директории, которая не копирует ваши зависимости. Проверьте пример внедрения в Airflow Github, вы поймете.
Во-первых, убедитесь, что структура файла в ZIP-формате соответствует ожиданиям.
Затем попробуйте получить путь:
os.path.join(os.path.dirname(os.path.realpath(__file__)),"sql","query_file.sql")
Вот как мы получаем пути к нашим запросам в нашем развертывании Airflow.