Airflow BashOperator：将参数传递给外部bash脚本

本文介绍了Airflow BashOperator：将参数传递给外部bash脚本的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

将参数从BashOperator传递到外部bash脚本时遇到问题。 运行本地命令时，参数已正确替换：

Having problems passing parameters to an external bash script from a BashOperator. When I run a local command, the params are substituted correctly:

log_cleanup = """ echo "{{ params.BASE_LOG_FOLDER }}" """ log_cleanup_task = BashOperator( task_id='log_cleanup_task', provide_context=True, bash_command = log_cleanup, params = {'BASE_LOG_FOLDER': "/var/opt"}, dag=dagInstance, ) prints: "/var/opt" (without the double quotes)

但是，如果我调用外部bash脚本，则参数不会替代。

But if I call an external bash script, the params don't substitute in.

log_cleanup_task = BashOperator( task_id='log_cleanup_task', provide_context=True, bash_command= str(DAGS_FOLDER)+"/scripts/log_cleanup.sh ", params = {'BASE_LOG_FOLDER': "/var/opt" }, dag=dagInstance, ) #log_cleanup.sh: #! /usr/bin/bash echo "{{ params.BASE_LOG_FOLDER }}" prints: "{{ params.BASE_LOG_FOLDER }}" (without the double quotes)

在外部bash脚本中，我无法像在语句中那样替换参数存储在DAG .py脚本中。

In the external bash script, I can't get the parameters to substitute in like they do when the statement is stored within the DAG .py script.

我是否必须将参数作为命令行参数传递？ Jinja模板仅适用于.py文件吗？

Do I have to pass the params as command line arguments instead? Does the jinja templating only works in the .py files?

推荐答案

删除 log_cleanup.sh 之后的空格c $ c> bash_command

Remove the space after "log_cleanup.sh " in bash_command

因此您的任务应变为：

log_cleanup_task = BashOperator( task_id='log_cleanup_task', provide_context=True, bash_command= "scripts/log_cleanup.sh", params = {'BASE_LOG_FOLDER': "/var/opt" }, dag=dagInstance, )

请注意，脚本文件夹应位于包含DAG文件的文件夹内，并且应包含脚本的相对路径（相对到包含此DAG的文件夹）

Note that the scripts folder should be inside the folder containing your DAG file and it should contain the relative path to script (relative to folder containing this DAG)

出现 TemplateNotFound 错误的主要原因是 bash_command 二手发动机y气流）。 Jinja仅识别在 DAG.template_searchpath 默认路径是包含DAG的文件夹，因此，如果DAG直接位于 $ AIRFLOW_HOME / dags中，则可以直接将脚本文件夹放置在DAGs文件夹下。或者，您可以按以下方式将路径传递到DAG.template_searchpath中的文件夹：

The main reason you got TemplateNotFound error was the path mentioned in bash_command is not recognized by Jinja (templating engine used by Airflow). Jinja only recognizes path passed in DAG.template_searchpath The default path is the folder containing the DAG so you can directly place your scripts folder under DAGs folder if your DAG is directly in the $AIRFLOW_HOME/dags. Or you can pass the path to your folder in DAG.template_searchpath as follows:

dag = DAG("example_dag", template_searchpath="/var/opt/scripts") # And then just pass "filename" to bash_command log_cleanup_task = BashOperator( task_id='log_cleanup_task', provide_context=True, bash_command= "log_cleanup.sh ", params = {'BASE_LOG_FOLDER': "/var/opt" }, dag=dag, )