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

Question

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

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 脚本，则参数不会替换。

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 脚本中存储语句时那样获取要替换的参数。

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

Answer 1

删除bash_command中"log_cleanup.sh "后面的空格

所以你的任务应该变成：

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,
)

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

您收到TemplateNotFound 错误的主要原因是bash_command 中提到的路径未被Jinja 识别（Airflow 使用的模板引擎）。 Jinja 只识别在DAG.template_searchpath 中传递的路径 默认路径是包含 DAG 的文件夹，因此如果您的 DAG 直接位于 $AIRFLOW_HOME/dags 中，您可以直接将脚本文件夹放在 DAGs 文件夹下。或者您可以将路径传递到 DAG.template_searchpath 中的文件夹，如下所示：

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,
)