如何使用 Airflow 在 python 中运行 bash 脚本答案

【问题标题】：How to run bash script in python using Airflow如何使用 Airflow 在 python 中运行 bash 脚本
【发布时间】：2021-11-08 13:06:46
【问题描述】：

我有一个python函数

def running_dump():
    with open('dags/scripts/shell_scripts/daily_pg_dump.sh', 'rb') as file:
        script = file.read()
        print(script)
    subprocess.call(script, shell=True)

和一个shell文件daily_pg_dump.sh

PGPASSWORD='*******' pg_dump -h ***** -p ***** -U ***** -d * -t table_1 > dags/data_bucket/table_1_backup.sql

气流夹

pg_dump_to_storage = PythonOperator(
        task_id='task_1',
        python_callable=running_dump,
        dag=dag
    )

当我使用 Airflow 调用 python 函数时，shell 脚本似乎没有运行，因为 table_1_backup.sql 没有创建。相反，我得到 Returned value was: 0 但没有出现错误。我错过了什么？

【问题讨论】：

Tagentially，你为什么要将脚本读入变量？作为二进制？你应该简单地subprocess.call(['dags/scripts/shell_scripts/daily_pg_dump.sh']) 虽然我不认为这会解决你的问题（我们也不能告诉你为什么它会在你的系统没有更多调试信息的情况下失败）。

标签： python bash airflow

【解决方案1】：

如果您想从 Airflow 运行 bash 脚本，您可以使用 BashOperator 而不是 PythonOperator。

从文档中的 example 来看，您的情况是：

from airflow.operators.bash import BashOperator

running_dump = “path/to/daily_pg_dump.sh ” # note the space after the script's name

pg_dump_to_storage = BashOperator(
   task_id='task_1', 
   bash_command=running_dump,
   dag=dag,
)

注意：当您在 bash 脚本中不使用 templating 时，您只需要在脚本路径名后面加上 space。

【讨论】：