AirflowのPostgresOperatorとJinjaのテンプレートとSQLを使用した場合

Question

PostgresOperatorでAirflowのテンプレート機能（Jinja2経由）を使用しようとすると、レンダリングすることができませんでした。私は間違ったことをしている可能性はかなり高いですが、問題の可能性についてはかなり失われています。

airflow_home = /home/gregreda/airflow 
dags_folder = /home/gregreda/airflow/dags 

関連DAGと変数

default_args = { 
    'owner': 'gregreda', 
    'start_date': datetime(2016, 6, 1), 
    'schedule_interval': None, 
    'depends_on_past': False, 
    'retries': 3, 
    'retry_delay': timedelta(minutes=5) 
} 

this_dag_path = '/home/gregreda/airflow/dags/example_csv_to_redshift' 
dag = DAG(
    dag_id='example_csv_to_redshift', 
    schedule_interval=None, 
    default_args=default_args 
) 

/example_csv_to_redshift/csv_to_redshift.py

airflow.cfg：ここで私はなってきたTemplateNotFoundエラーを再現する例を示します。

を呼び出す

/example_csv_to_redshift/copy_to_redshift.sql

COPY public.table_foobar FROM 's3://mybucket/test-data/import/foobar.csv' 
CREDENTIALS 'aws_access_key_id={{ AWS_ACCESS_KEY_ID }};aws_secret_access_key={{ AWS_SECRET_ACCESS_KEY }}' 
CSV 
NULL as 'null' 
IGNOREHEADER as 1; 

は、以下の例外がスローされます。注：別のDAGについてもこの問題が発生しています。そのため、example_redshift_query_to_csvのパスが表示されます。

[2016-06-14 21:24:57,484] {__init__.py:36} INFO - Using executor SequentialExecutor 
[2016-06-14 21:24:57,565] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python2.7/lib2to3/Grammar.txt 
[2016-06-14 21:24:57,596] {driver.py:120} INFO - Generating grammar tables from /usr/lib/python2.7/lib2to3/PatternGrammar.txt 
[2016-06-14 21:24:57,763] {models.py:154} INFO - Filling up the DagBag from /home/gregreda/airflow/dags 
[2016-06-14 21:24:57,828] {models.py:2040} ERROR - /home/gregreda/airflow/dags/example_redshift_query_to_csv/export_query_to_s3.sql 
Traceback (most recent call last): 
    File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 2038, in resolve_template_files 
    setattr(self, attr, env.loader.get_source(env, content)[0]) 
    File "/usr/local/lib/python2.7/dist-packages/jinja2/loaders.py", line 187, in get_source 
    raise TemplateNotFound(template) 
TemplateNotFound: /home/gregreda/airflow/dags/example_redshift_query_to_csv/export_query_to_s3.sql 
[2016-06-14 21:24:57,834] {models.py:2040} ERROR - /home/gregreda/airflow/dags/example_csv_to_redshift/copy_to_redshift.sql 
Traceback (most recent call last): 
    File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 2038, in resolve_template_files 
    setattr(self, attr, env.loader.get_source(env, content)[0]) 
    File "/usr/local/lib/python2.7/dist-packages/jinja2/loaders.py", line 187, in get_source 
    raise TemplateNotFound(template) 
TemplateNotFound: /home/gregreda/airflow/dags/example_csv_to_redshift/copy_to_redshift.sql 
Traceback (most recent call last): 
    File "/usr/local/bin/airflow", line 15, in <module> 
    args.func(args) 
    File "/usr/local/lib/python2.7/dist-packages/airflow/bin/cli.py", line 359, in render 
    ti.render_templates() 
    File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 1409, in render_templates 
    rendered_content = rt(attr, content, jinja_context) 
    File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line 2017, in render_template 
    return jinja_env.get_template(content).render(**context) 
    File "/usr/local/lib/python2.7/dist-packages/jinja2/environment.py", line 812, in get_template 
    return self._load_template(name, self.make_globals(globals)) 
    File "/usr/local/lib/python2.7/dist-packages/jinja2/environment.py", line 774, in _load_template 
    cache_key = self.loader.get_source(self, name)[1] 
    File "/usr/local/lib/python2.7/dist-packages/jinja2/loaders.py", line 187, in get_source 
    raise TemplateNotFound(template) 
jinja2.exceptions.TemplateNotFound: /home/gregreda/airflow/dags/example_csv_to_redshift/copy_to_redshift.sql 

修正のためのアイデアは大変ありがたいです。

Answer 1

標準PEBCAK error。

与えられたエアフロータスク内のSQLテンプレートへのパスを指定する際に、相対パスで指定する必要がある問題がありました。さらに

copy_s3_to_redshift = PostgresOperator(
    task_id='load_table', 
    sql='/copy_to_redshift.sql', 
    params=dict(
     AWS_ACCESS_KEY_ID=Variable.get('AWS_ACCESS_KEY_ID'), 
     AWS_SECRET_ACCESS_KEY=Variable.get('AWS_SECRET_ACCESS_KEY') 
    ), 
    postgres_conn_id='postgres_redshift', 
    autocommit=False, 
    dag=dag 
) 

、若干変更する必要がSQLテンプレート（params. ...この時に注意してください）：もう少しコントロールのために

COPY public.pitches FROM 's3://mybucket/test-data/import/heyward.csv' 
CREDENTIALS 'aws_access_key_id={{ params.AWS_ACCESS_KEY_ID }};aws_secret_access_key={{ params.AWS_SECRET_ACCESS_KEY }}' 
CSV 
NULL as 'null' 
IGNOREHEADER as 1; 

Answer 2

、template_searchpathのparamを使用してDAGを初期化、そしてちょうど使用オペレータのファイル名。

:param template_searchpath: This list of folders (non relative) 
    defines where jinja will look for your templates. Order matters. 
    Note that jinja/airflow includes the path of your DAG file by 
    default 
:type template_searchpath: string or list of stings