使用 Numpy Pickle 读取 Joblib 保存文件的 Celery 问题

Question

Celery 尝试读取我们使用 joblib 保存的模型文件时遇到了一个非常奇怪的问题。这是一个使用工厂模式的 Flask 应用程序。这是整体设置：

fit_model.py:

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder
import joblib


class MyClass(Pipeline):
    pass


if __name__ == "__main__":
    to_persist = MyClass(steps=[("step", OneHotEncoder())])
    joblib.dump(to_persist, "dummy.model")

load_model.py:

import joblib


if __name__ == "__main__":
    with open("dummy.model", "rb") as f:
        my_object = joblib.load(f)
        assert my_object is not None

tasks.py:

import joblib

@celery_app.task()
def dummy_task():
    _ = joblib.load("dummy.model")

如果我们运行lit_model.py 文件，然后运行load_model.py，它工作得非常好。但是，当它作为 Celery 任务运行并且我们尝试延迟 dummy_task 时，我们会得到以下信息：

  File "/usr/local/lib/python3.7/site-packages/celery/app/trace.py", line 412, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/celery/app/trace.py", line 704, in __protected_call__
    return self.run(*args, **kwargs)
  File "/lit_service/tasks.py", line 32, in dummy_task
    _ = joblib.load("dummy.model")
  File "/usr/local/lib/python3.7/site-packages/joblib/numpy_pickle.py", line 585, in load
    obj = _unpickle(fobj, filename, mmap_mode)
  File "/usr/local/lib/python3.7/site-packages/joblib/numpy_pickle.py", line 504, in _unpickle
    obj = unpickler.load()
  File "/usr/local/lib/python3.7/pickle.py", line 1088, in load
    dispatch[key[0]](self)
  File "/usr/local/lib/python3.7/pickle.py", line 1376, in load_global
    klass = self.find_class(module, name)
  File "/usr/local/lib/python3.7/pickle.py", line 1430, in find_class
    return getattr(sys.modules[module], name)
AttributeError: module 'celery.bin.celery' has no attribute 'MyClass'

Flask Celery worker 在application.py 中的设置如下：

celery_app = Celery(
    __name__, broker=settings.CELERY_BROKER_URL, backend=settings.CELERY_BROKER_URL
)

def create_app(**config_overrides):
    app = Flask(__name__)

    # start celery
    celery_app.conf.update(app.config)
    celery_app.conf.imports = ["tasks"]
    celery_app.conf.task_routes = {"tasks.*": {"queue": "lit_tasks"}}

    return app

worker 生成为 Docker 容器，如下所示：

celery worker -A celery_worker.celery_app --loglevel=INFO --queues lit_tasks

关于在哪里寻找的任何线索？

Answer 1

尝试在自己的文件中定义MyClass，如my_class.py，并将from my_class import MyClass添加到fit_model.py中。