如何延伸Python

Question

我正试图将我的PythonOperator调焦，并将其放置在$AIRFLOW\u HOME/plugins下面，如下图所示：

class MyPythonOperator(PythonOperator):

    def my_callable(param1, param2, param3):
        # do something

    @apply_defaults
    def __init__(self, task_id, *args, **kwargs):

        super(MyPythonOperator, self).__init__(
            task_id=task_id,  
            python_callable = self.my_callable,
            provide_context = True,
            *args, **kwargs)

然后我定义了一个气流dag代码，它非常简单，只完成两个任务：

^{pr2}$

但是，在运行dag python代码后，获取错误消息：

^{pr3}$

我尝试了一些调试，并在此行插入断点：

^{cd1>}

在调用超级构造函数之前，我发现自耗以及自我.dag\u id异常，值为：

^{pr4}$

有人知道这棵枫树有什么问题吗？ 非常感谢你！

Answer 1

我最近也碰到过这个。似乎您缺少自定义PythonOperator上的context参数。在

更改方法定义，使其如下所示：

def my_callable(param1, param2, param3, **context):
    # do something

失败的原因是您在操作符中提供的provide_context=True标志。出于某种原因，python可调用函数正在您的params中查找它。在