扩展Airflow BashOperator自定义ROperator报错：缺少bash_command参数

问题原因 #

BashOperator的构造方法要求必须传入bash_command参数，这是它的核心必填参数——哪怕你重写了execute方法，父类的初始化逻辑仍然会检查这个参数是否存在。你当前的代码调用super().__init__(*args, **kwargs)时没有传递该参数，因此Airflow抛出了缺失关键字参数的异常。

修复方案 #

在自定义Operator的__init__方法中，先构造好要执行的R命令，将其作为bash_command参数传给父类的构造方法，满足父类的参数要求。同时可以简化execute方法，直接复用父类的逻辑（或者保留自定义逻辑，只要父类初始化时拿到合法的bash_command即可）。

修正后的完整代码 #

from airflow.operators.bash_operator import BashOperator
from airflow.utils.decorators import apply_defaults
from airflow.exceptions import AirflowException  # 补充导入缺失的异常类

class ROperator(BashOperator):
    """
    Execute an R script.

    :param script_path: The local path to the R script.
    :type script_path: str
    :param args: List of arguments to pass to the R script.
    :type args: list
    :param r_cmd: The command to use to invoke R. The default is
        ``"Rscript"``, which should work in most cases.
    :type r_cmd: str
    """

    @apply_defaults
    def __init__(
        self,
        script_path,
        args=None,
        r_cmd="Rscript",
        *args,
        **kwargs
    ):
        # 构造完整的R执行命令，作为bash_command传给父类
        bash_command = f"{r_cmd} {script_path}"
        if args:
            bash_command += " " + " ".join(args)
        
        # 将bash_command传入父类初始化，满足BashOperator的参数要求
        super().__init__(bash_command=bash_command, *args, **kwargs)
        
        self.script_path = script_path
        self.args = args or []
        self.r_cmd = r_cmd

    def execute(self, context):
        # 直接复用父类的execute逻辑，无需自己调用execute_command
        return_code = super().execute(context)
        if return_code != 0:
            raise AirflowException("R脚本执行失败")

额外说明 #

1. 修正了原代码中的参数名不一致问题：文档注释写的是args，但代码里用的是arg，统一为args更规范。
2. 补充导入了AirflowException，原代码中抛出异常时未导入该类，会导致另一个报错。
3. 类名修正为ROperator（原代码是ROpertor，少了一个字母p），符合命名规范。

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