Airflow BashOperator调用Python脚本DAG无执行效果及路径报错问题

问题根源分析 #

1. DAG任务未实际执行脚本：原代码中用@task装饰器包裹函数，但函数内仅实例化BashOperator却未返回或触发执行，导致Airflow认为任务完成，但实际没有运行任何逻辑。
2. 脚本路径配置错误：路径调整后出现文件找不到的报错，是因为Airflow模板渲染机制会将脚本临时复制到tmp目录，但template_searchpath未正确指向脚本目录，或路径写法不符合Airflow模板规则；同时Python脚本中使用的相对路径基于Airflow任务工作目录（非DAG所在目录），导致数据文件无法定位。

具体解决方案 #

一、修复DAG任务的写法错误 #

方法1：直接使用BashOperator（推荐）

移除冗余的@task装饰器，直接实例化BashOperator作为任务节点，并配置正确的模板搜索路径：

import json
from pendulum import datetime

from airflow.operators.bash import BashOperator
from airflow.models.baseoperator import chain
from airflow.decorators import dag

# 脚本目录相对于DAG文件的路径，或使用绝对路径
PYTHON_SCRIPTS_DIR = "./python_scripts"

@dag(
    schedule="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False,
    default_args={"retries": 2},
    # 设置模板搜索路径，让Airflow能找到脚本文件
    template_searchpath=PYTHON_SCRIPTS_DIR
)
def parse_json_data():
    parse_json_task = BashOperator(
        task_id="parse_json_task",
        # 使用Airflow模板变量引用脚本，确保模板渲染正常
        bash_command="python {{ templates_dir }}/json_file_parser.py"
    )

    load_files_task = BashOperator(
        task_id="load_files_task",
        bash_command="python {{ templates_dir }}/load_data.py"
    )

    chain(parse_json_task, load_files_task)

parse_json_data()

方法2：使用@task.bash装饰器简化写法

如果偏好装饰器风格，直接用@task.bash替代普通@task，无需手动实例化BashOperator：

import json
from pendulum import datetime

from airflow.models.baseoperator import chain
from airflow.decorators import dag, task

PYTHON_SCRIPTS_DIR = "./python_scripts"

@dag(
    schedule="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False,
    default_args={"retries": 2},
    template_searchpath=PYTHON_SCRIPTS_DIR
)
def parse_json_data():
    @task.bash
    def parse_json():
        # 返回bash命令，利用模板变量定位脚本
        return "python {{ templates_dir }}/json_file_parser.py"

    @task.bash
    def load_files():
        return "python {{ templates_dir }}/load_data.py"

    chain(parse_json(), load_files())

parse_json_data()

二、修复脚本路径与依赖问题 #

1. 规范脚本存放路径

   • 确保python_scripts目录与DAG文件处于同一目录下，或template_searchpath设置为脚本目录的绝对路径（如/opt/airflow/dags/python_scripts）；
   • 检查脚本文件权限，确保Airflow运行用户拥有读取权限。

2. 修正Python脚本中的相对路径
   脚本中../data_files/xxx这类相对路径会因Airflow工作目录差异失效，可通过两种方式修复：

   • 使用绝对路径：直接替换为数据文件的绝对路径，例如：

     input_json_file = "/opt/airflow/data_files/json_file.jsonl.gz"
     parsed_json_file = "/opt/airflow/data_files/parsed_json_file.json"
     

   • 通过环境变量传递路径：在DAG的BashOperator中添加环境变量，脚本读取该变量拼接路径：
     DAG中修改BashOperator：

     BashOperator(
         task_id="parse_json_task",
         bash_command="python {{ templates_dir }}/json_file_parser.py",
         env={"DATA_DIR": "/opt/airflow/data_files"}
     )
     

     脚本中调整：

     import os
     DATA_DIR = os.getenv("DATA_DIR")
     input_json_file = os.path.join(DATA_DIR, "json_file.jsonl.gz")
     parsed_json_file = os.path.join(DATA_DIR, "parsed_json_file.json")
     

3. 修复load_data.py中的函数名错误
   脚本main()方法中调用的load_processed_nhtsa_file和load_nhtsa_lookup_file与定义的函数名load_processed_json_file、load_lookup_file不匹配，需修正为一致，否则会触发函数未定义错误。

三、额外排查步骤 #

• 查看Airflow任务日志：在UI中点击任务实例，查看详细日志，确认脚本执行情况及具体报错；
• 验证Python环境：确保Airflow使用的Python环境已安装gzip、pandas、sqlalchemy等依赖包；
• 检查临时目录权限：确保Airflow运行用户有权限读写tmp目录（报错中的/private/var/folders/...路径）。

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