能否在Cloud Composer的DAG中执行独立Python脚本，应选用哪种Operator？

首先明确答案：可以将你现有的长Python脚本放到Cloud Composer的任务中执行，两种Operator都支持，不过你给出的PythonOperator示例写法是错误的，python_callable参数只能接收Python可调用对象（比如函数），不能直接传入.py文件路径，无法直接运行。

方案1：使用PythonOperator执行脚本逻辑 #

适用场景 #

• 需要和Airflow上下文交互（比如获取执行日期、传递XCom参数、使用Airflow内置的变量/连接配置）
• 脚本没有和Airflow运行环境冲突的特殊依赖，不需要单独的隔离运行环境
• 可以对现有脚本做少量代码调整

操作方法 #

你可以先在Cloud Composer对应的GCS dags 目录下新建一个scripts文件夹，把所有Python脚本放到这个目录下，Cloud Composer会自动把dags目录加到Python的模块搜索路径中。
在你的Python脚本中新增一个入口函数，把原有脚本逻辑都封装到这个函数里：

# myscript1.py 示例内容
def run_script(**context):
    # 此处放你原有脚本的全部逻辑
    print("执行myscript1的业务逻辑")
    # 有返回值可以自动存到XCom中供下游任务调用
    return "执行完成"

之后在DAG文件中按如下方式编写即可：

from scripts import myscript1, myscript2
import datetime
from airflow import models
from airflow.operators import python_operator

default_dag_args = {}
with models.DAG(
    'jobname',
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args
) as dag: 
    do_stuff1 = python_operator.PythonOperator(
        task_id='task_1',
        python_callable=myscript1.run_script,
        provide_context=True # 需要获取Airflow上下文时加这个参数
    )
    do_stuff2 = python_operator.PythonOperator(
        task_id='task_2',
        python_callable=myscript2.run_script
    )

优势 #

• 脚本运行日志直接输出到Airflow任务日志中，排查问题方便
• 可以直接使用Airflow的所有内置能力，参数传递、数据互通不需要额外写适配逻辑
• 运行异常会被Airflow直接捕获，任务状态判定更准确

方案2：使用BashOperator调用Python脚本 #

适用场景 #

• 你的脚本有特殊依赖，和Airflow运行环境的依赖版本冲突，需要指定独立的Python解释器运行
• 脚本本身是可独立运行的完整程序，不需要和Airflow上下文交互
• 不想修改原有脚本的代码结构，希望直接运行原文件

操作方法 #

同样把脚本放到dags/scripts目录下，DAG文件按如下方式编写：

import datetime
from airflow import models
from airflow.operators import bash_operator

default_dag_args = {}
with models.DAG(
    'jobname',
    schedule_interval=datetime.timedelta(days=1),
    default_args=default_dag_args
) as dag: 
    do_stuff1 = bash_operator.BashOperator(
        task_id='task_1',
        bash_command='python3 $AIRFLOW_HOME/dags/scripts/myscript1.py'
    )
    do_stuff2 = bash_operator.BashOperator(
        task_id='task_2',
        # 需要传递Airflow参数可以直接用模板变量，比如传执行日期加{{ ds }}即可
        bash_command='python3 $AIRFLOW_HOME/dags/scripts/myscript2.py {{ ds }}'
    )

注意事项 #

• 你需要保证脚本运行正常时退出码为0，异常时返回非0退出码，Airflow会根据退出码判定任务是否成功
• 脚本的打印输出也会同步到Airflow任务日志中，但是如果要把数据传给下游任务，需要自己做额外的持久化处理

推荐选择 #

绝大多数场景下更推荐使用PythonOperator的方案，只有当你确实有依赖隔离、不想修改原有脚本的需求时，再选择BashOperator方案。

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