Geek Answers Handbook

TemplateNotFound when using Airflow PostgresOperator with Jinja template and SQL

When I tried to use the Templating features for Airflow (via Jinja2) using PostgresOperator, I was unable to get things to render. It is possible that I am doing something wrong, but I pretty much lost what might be the problem. Here is an example to reproduce the TemplateNotFound error I was getting:

airflow.cfg

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

corresponding dag and variables

 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

 copy_s3_to_redshift = PostgresOperator( task_id='load_table', sql=this_dag_path + '/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 )

/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;

Calling airflow render example_csv_to_redshift load_table 2016-06-14 raises the following exception. Note. I also encounter this problem for another DAG, so you see the path with the specified 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

Any ideas for a fix are greatly appreciated.

+5
source share
2 answers

PEBCAK Standard Error .

There was a problem indicating the path to the SQL template in the Airflow task, which should be relative.

 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 )

In addition, you need to slightly modify the SQL template (pay attention to params. ... this time):

 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;
+3
source

For a little more control, run your DAG with the template_searchpath parameter, and then just use the file name in the statement.

 :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
0
source

More articles:


All Articles