Airflow 异常.AirflowException : Use keyword arguments when initializing operators

Question

我在 Ubuntu 中使用 Airflow 版本 1.9.2 和 Python 2.7。我尝试从 ParentOperator 类继承，该类本身工作正常，并创建一个名为 ChildOperator 的类。但是当我创建一个 ChildOperator 实例时，我认为这里缺少或弄乱了一些关键字参数，并且我收到此错误:

airflow.exceptions.AirflowException: Use keyword arguments when initializing operators

这是一个简化的示例:

class ParentOperator(BaseOperator, SkipMixin):

    @apply_defaults
    def __init__(self,
             conn_id,
             object,
             args={},
             s3_conn_id=None,
             s3_key=None,
             s3_bucket=None,
             fields=None,
             *args,
             **kwargs
             ):

        super(ParentOperator, self).__init__(*args, **kwargs)
        ...


class ChildOperator(ParentOperator):

    @apply_defaults
    def __init__(self,
             conn_id,
             object,
             args={},
             s3_conn_id=None,
             s3_key=None,
             s3_bucket=None,
             fields=None,
             *args,
             **kwargs
             ):

        args=...
        super(ChildOperator, self).__init__(
            conn_id,
            object,
            args=args,
            s3_conn_id=s3_conn_id,
            s3_key=s3_key,
            s3_bucket=s3_bucket,
            fields=fields,
            *args,
            **kwargs
        )

...

myobjc = ChildOperator(
    conn_id="my_default",
    object=table,
    args={},
    s3_conn_id='s3_postgres_dump',
    s3_key=s3_key,
    s3_bucket=s3_bucket,
    dag=dag,
    task_id="task1"
)

知道是什么导致了这个错误吗？这更像是 Python 特有的问题吗？

Answer 1

ChildOperator 的

__init__ 函数需要具有如下所示的所有关键字参数(对于 conn_id 和 object 的前两个参数):

super(ChildOperator, self).__init__(
        conn_id=conn_id,
        object=object,
        args=args,
        s3_conn_id=s3_conn_id,
        s3_key=s3_key,
        s3_bucket=s3_bucket,
        fields=fields,
        *args,
        **kwargs
)