Cloud Composer(Airflow)调用依赖Python脚本的方法咨询

完全可行，你完全可以在DAG里只声明单个入口脚本（通过BashOperator/PythonOperator调用），再由这个入口脚本去调用项目里的其他Python脚本，不需要把所有逻辑都直接堆在DAG文件里，只要做好配置避开常见坑即可。

针对你给出的BashOperator配置的修正说明 #

你写的配置直接传bash_command="test.py"默认是跑不通的，需要补全以下必要配置：

• Cloud Composer会自动把GCS存储桶里dags路径下的所有文件同步到所有Worker节点的/home/airflow/gcs/dags/目录，也就是Airflow内置的DAGS_FOLDER路径，所有你需要调用的py脚本（包括入口test.py和它要调用的其他业务脚本）都必须放在这个DAG目录或其子目录下，否则Worker会找不到文件。
• 执行命令时最好直接指定Python解释器+脚本的绝对路径，不用额外给脚本加可执行权限、写shebang，最稳妥的写法是直接引用环境变量拼路径：bash_command="python3 $DAGS_FOLDER/test.py"。

入口脚本调用其他Python脚本的注意事项 #

• 不要在test.py里用相对路径找其他脚本，不同Worker节点执行任务时的工作目录不固定，最好在test.py里通过DAGS_FOLDER环境变量拼接所有被调用脚本的绝对路径，路径拼接参考示例：

import os
import subprocess

DAG_PATH = os.environ.get("DAGS_FOLDER")
# 调用同目录下的other_script.py
other_script_path = os.path.join(DAG_PATH, "other_script.py")
# 执行子脚本，加check=True保证子脚本报错时主任务能感知到失败
subprocess.run(["python3", other_script_path], check=True)

• 如果是用PythonOperator调入口逻辑，入口脚本里导入其他自定义Python模块也是完全支持的，跨目录导入时可以在入口脚本开头把DAGS_FOLDER加到sys.path里即可。
• 所有脚本依赖的第三方Python包，提前在Cloud Composer环境的PyPI包管理页面安装好，不要在任务运行时临时执行pip install，会导致任务不稳定、偶发失败。
• 不要把业务脚本放在DAG目录以外的本地路径，Cloud Composer的Worker是弹性扩缩容的，本地存储的文件不会在多节点间同步，换节点执行就会报文件不存在的错误。

修正后的BashOperator示例代码 #

curl = BashOperator(
    task_id='multiplescripts',
    bash_command="python3 $DAGS_FOLDER/test.py",
    dag=dag
)

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