在尝试将 Airflow 的模板功能(通过 Jinja2)与 PostgresOperator 一起使用时,我一直无法渲染内容。我很可能做错了什么,但我对问题可能是什么感到很迷茫。这是一个重现我遇到的 TemplateNotFound 错误的示例:
气流.cfg
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
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;
调用airflow render example_csv_to_redshift load_table 2016-06-14
会引发以下异常。请注意,对于另一个 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
任何关于修复的想法都非常感谢。