Airflow中BashOperator的bash_command未实际执行问题求助

问题描述 #

你编写的Airflow DAG代码如下：

from datetime import datetime
from airflow.decorators import task,dag
from airflow.operators.python import PythonOperator
from airflow.operators.bash import BashOperator

args = {
    'owner': 'Harry',
    'depends_on_past': True,
    'start_date': datetime(2022,1,1)
}
    
@dag(default_args=args,schedule_interval="@once",description='Task da runcode do professor',catchup=False,tags=['runcode'])
    def runcode():
    
        @task
        def start():
            print('inicio da carga dos dados.')
        
        @task
        def cria_parquet():
            BashOperator(
            task_id = 'chama arquivo',
            bash_command='/usr/local/bin/python3.7 /opt/airflow/dags/prova.py'
        )
    
        instancia_inicio = start()
        instancia_cria_parquet = cria_parquet()
    
        instancia_inicio >> instancia_cria_parquet
    
execucao = runcode()

遇到的问题：在@task装饰的Python任务中实例化BashOperator，指定bash_command调用外部Python脚本prova.py，任务执行后显示“成功”，但脚本内的代码并未实际运行；直接在bash终端执行该命令时可正常运行。

原因分析 #

核心问题是：在@task装饰的Python函数里仅仅实例化BashOperator完全没用。

@task装饰的函数本质是一个Python任务，它只会执行函数内部的Python代码逻辑——你只是创建了BashOperator对象，但根本没有触发它的执行流程。Airflow的Operator是独立的任务单元，需要被纳入DAG的任务依赖链中作为节点，或者显式调用其执行方法，否则它就是个普通的Python对象，不会实际运行对应的bash命令。

修正方案 #

有两种正确的处理方式，推荐第一种符合Airflow设计逻辑的用法：

方式1：将BashOperator作为独立任务，不嵌套在@task中 #

把cria_parquet改成直接定义BashOperator实例，作为DAG的独立任务节点：

from datetime import datetime
from airflow.decorators import task, dag
from airflow.operators.bash import BashOperator

args = {
    'owner': 'Harry',
    'depends_on_past': True,
    'start_date': datetime(2022,1,1)
}
    
@dag(default_args=args,schedule_interval="@once",description='Task da runcode do professor',catchup=False,tags=['runcode'])
def runcode():
    @task
    def start():
        print('inicio da carga dos dados.')
    
    # 直接定义BashOperator作为独立任务
    cria_parquet_task = BashOperator(
        task_id='chama_arquivo',
        bash_command='/usr/local/bin/python3.7 /opt/airflow/dags/prova.py'
    )
    
    # 构建任务依赖
    start() >> cria_parquet_task

execucao = runcode()

方式2：在Python任务中直接执行bash命令（不推荐嵌套Operator） #

如果一定要在Python任务里执行bash命令，直接用subprocess模块调用，不要实例化BashOperator：

from datetime import datetime
from airflow.decorators import task, dag
import subprocess

args = {
    'owner': 'Harry',
    'depends_on_past': True,
    'start_date': datetime(2022,1,1)
}
    
@dag(default_args=args,schedule_interval="@once",description='Task da runcode do professor',catchup=False,tags=['runcode'])
def runcode():
    @task
    def start():
        print('inicio da carga dos dados.')
    
    @task
    def cria_parquet():
        # 用subprocess直接执行bash命令
        result = subprocess.run(
            ['/usr/local/bin/python3.7', '/opt/airflow/dags/prova.py'],
            capture_output=True,
            text=True
        )
        # 打印输出并处理异常
        print(result.stdout)
        if result.returncode != 0:
            raise Exception(f"脚本执行失败: {result.stderr}")
    
    start() >> cria_parquet()

execucao = runcode()

总结 #

Airflow的Operator是独立的任务单元，要么直接作为DAG的任务节点使用，要么不要在Python任务里实例化它——嵌套实例化只会创建对象，不会触发实际执行。优先使用方式1，这是Airflow的标准用法，更便于监控每个任务的状态和调试。

内容的提问来源于stack exchange，提问作者Astora