运行简单 Airflow BashOperator 时出现 TemplateNotFound 错误

Posted 2023-02-16

【中文标题】运行简单 Airflow BashOperator 时出现 TemplateNotFound 错误

【英文标题】：TemplateNotFound error when running simple Airflow BashOperator

【发布时间】：2017-06-28 02:25:29

【问题描述】：

我正在尝试编写我们的第一个 Airflow DAG，当我尝试使用命令 airflow list_tasks orderwarehouse 列出任务时出现以下错误：

Traceback (most recent call last):
  File "/usr/local/lib/python2.7/site-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/site-packages/jinja2/loaders.py", line 187, in get_source
    raise TemplateNotFound(template)
TemplateNotFound: ./home/deploy/airflow-server/task_scripts/orderwarehouse/load_warehouse_tables.sh

此 DAG 不应使用模板。我只是尝试按照the docs 中的说明在指定位置运行 shell 脚本。 shell 脚本确实存在于该位置并且拼写正确。我的 DAG 如下所示：

from airflow import DAG
from airflow.operators.bash_operator import BashOperator

default_args = 
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2015, 6, 1),
    'email': ['[e-mail]'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),


orderwarehouse = DAG('orderwarehouse', default_args=default_args)

load_mysql = BashOperator(
    task_id='load_warehouse_mysql',
    bash_command='./home/deploy/airflow-server/task_scripts/orderwarehouse/load_warehouse_tables.sh',
    dag=orderwarehouse)

不确定为什么它认为需要寻找 Jinja 模板。没有关于这个的想法，如果有人能指出我误入歧途的地方，我将不胜感激。谢谢。

【参考方案1】：

这是气流的一个陷阱。在 bash_command 末尾添加一个空格，它应该可以正常运行

来源： https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=62694614

【讨论】：

发现它也适用于文件名前的换行符！

如果命令后面有参数，这是否也适用（例如bash /path/to/script.sh arg1 arg2 ...）？文档似乎没有说明。

我认为可以，但我还没有测试过。

2020年了，怎么还会有这个bug？我只是花了几个小时来追踪这个。 :(

现在是 2021 年了，漏洞仍然存在

【参考方案2】：

除了提供的所有答案之外，我还必须做更多的事情来摆脱 jinja template not found 问题。

我们必须在任务定义中的文件名后添加空格。

【参考方案3】：

您应该尝试在文件路径末尾添加空格。无论您使用哪个运算符，都应始终遵循相同的规则。

load_mysql = BashOperator(  
             task_id='load_warehouse_mysql',  
             command='/home/deploy/airflow-server/task_scripts/orderwarehouse/load_warehouse_tables.sh ',  
             dag=orderwarehouse)