Celery：访问链中的所有先前结果

Question

所以基本上我有一个相当复杂的工作流程，看起来类似于：

>>> res = (add.si(2, 2) | add.s(4) | add.s(8))()
>>> res.get()
16

之后，沿着结果链向上走并收集所有单独的结果对我来说相当微不足道：

>>> res.parent.get()
8

>>> res.parent.parent.get()
4

我的问题是，如果我的第三个任务取决于知道第一个任务的结果，但在这个例子中只收到第二个的结果怎么办？

而且链很长，结果也不是那么小，所以仅仅通过输入作为结果会不必要地污染结果存储。这是 Redis，因此使用 RabbitMQ、ZeroMQ 时的限制不适用。

Answer 1

一个简单的解决方法是将任务结果存储在列表中并在您的任务中使用它们。

from celery import Celery, chain
from celery.signals import task_success

results = []

app = Celery('tasks', backend='amqp', broker='amqp://')


@task_success.connect()
def store_result(**kwargs):
    sender = kwargs.pop('sender')
    result = kwargs.pop('result')
    results.append((sender.name, result))


@app.task
def add(x, y):
    print("previous results", results)
    return x + y

现在，在你的链中，所有以前的结果都可以从任何任务以任何顺序访问。

Answer 2

我为每个链分配一个作业 ID，并通过将数据保存在数据库中来跟踪该作业。

正在启动队列

if __name__ == "__main__":
  # Generate unique id for the job
  job_id = uuid.uuid4().hex
  # This is the root parent
  parent_level = 1
  # Pack the data. The last value is your value to add
  parameters = job_id, parent_level, 2
  # Build the chain. I added an clean task that removes the data
  # created during the process (if you want it)
  add_chain = add.s(parameters, 2) | add.s(4) | add.s(8)| clean.s()
  add_chain.apply_async()

现在任务

#Function for store the result. I used sqlalchemy (mysql) but you can
# change it for whatever you want (distributed file system for example)
@inject.params(entity_manager=EntityManager)
def save_result(job_id, level, result, entity_manager):
  r = Result()
  r.job_id = job_id
  r.level = level
  r.result = result
  entity_manager.add(r)
  entity_manager.commit()

#Restore a result from one parent
@inject.params(entity_manager=EntityManager)
def get_result(job_id, level, entity_manager):
  result = entity_manager.query(Result).filter_by(job_id=job_id, level=level).one()
  return result.result

#Clear the data or do something with the final result
@inject.params(entity_manager=EntityManager)
  def clear(job_id, entity_manager):
  entity_manager.query(Result).filter_by(job_id=job_id).delete()

@app.task()
def add(parameters, number):
  # Extract data from parameters list
  job_id, level, other_number = parameters

  #Load result from your second parent (level - 2)
  #For level 3 parent level - 3 and so on
  #second_parent_result = get_result(job_id, level - 2)

  # do your stuff, I guess you want to add numbers
  result = number + other_number
  save_result(job_id, level, result)

  #Return the result of the sum or anything you want, but you have to send something because the "add" function expects 3 values
  #Of course your should return the actual job and increment the parent level
  return job_id, level + 1, result

@app.task()
def clean(parameters):
  job_id, level, result = parameters
  #Do something with final result or not
  #Clear the data
  clear(job_id)

我使用 entity_manager 来管理数据库操作。我的实体管理器使用 sql 炼金术和 mysql。我还使用 table "result" 来存储部分结果。这部分应该更改为您最好的存储系统（或者如果 mysql 适合您就使用它）

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
import inject

class EntityManager():

  session = None

  @inject.params(config=Configuration)
  def __init__(self, config):
    conf = config['persistence']
    uri = conf['driver'] + "://" + conf['username'] + ":@" + conf['host'] + "/" + conf['database']

    engine = create_engine(uri, echo=conf['debug'])

    Session = sessionmaker(bind=engine)
    self.session = Session()

  def query(self, entity_type):
    return self.session.query(entity_type)

  def add(self, entity):
    return self.session.add(entity)

  def flush(self):
    return self.session.flush()

  def commit(self):
    return self.session.commit()

class Configuration:
  def __init__(self, params):
    f = open(os.environ.get('PYTHONPATH') + '/conf/config.yml')
    self.configMap = yaml.safe_load(f)
    f.close()

  def __getitem__(self, key: str):
    return self.configMap[key]

class Result(Base):
  __tablename__ = 'result'

  id = Column(Integer, primary_key=True)
  job_id = Column(String(255))
  level = Column(Integer)
  result = Column(Integer)

  def __repr__(self):
    return "<Result (job='%s', level='%s', result='%s')>" % (self.job_id, str(self.level), str(self.result))

我使用包注入来获取依赖注入器。注入包将重用该对象，因此您可以在每次需要时注入对数据库的访问，而不必担心连接问题。

class配置是在配置文件中加载数据库访问数据。您可以替换它并使用静态数据（硬编码的地图）进行测试。

为您更改任何其他 suitable 的依赖项注入。这只是我的解决方案。我只是为了快速测试添加它。

这里的关键是将部分结果保存在队列系统的某个地方，并在任务中 return 访问这些结果的数据（job_id 和父级别）。您将发送这个额外的（但很小的）数据，它是一个指向真实数据（一些大东西）的地址（job_id + 父级别）。

这个解决方案就是我在我的软件中使用的解决方案

Answer 3

也许您的设置对于这个来说太复杂了，但我喜欢使用 group 结合 noop 任务来完成类似的事情。我这样做是因为我想突出显示在我的管道中仍然同步的区域（通常这样它们就可以被删除）。

使用与您的示例类似的东西，我从一组看起来像这样的任务开始：

tasks.py:

from celery import Celery

app = Celery('tasks', backend="redis", broker='redis://localhost')


@app.task
def add(x, y):
        return x + y


@app.task
def xsum(elements):
    return sum(elements)


@app.task
def noop(ignored):
    return ignored

通过这些任务，我创建了一个链，使用一个组来控制依赖于同步结果的结果：

In [1]: from tasks import add,xsum,noop
In [2]: from celery import group

# First I run the task which I need the value of later, then I send that result to a group where the first task does nothing and the other tasks are my pipeline.
In [3]: ~(add.si(2, 2) | group(noop.s(),  add.s(4) | add.s(8)))
Out[3]: [4, 16]

# At this point I have a list where the first element is the result of my original task and the second element has the result of my workflow.
In [4]: ~(add.si(2, 2) | group(noop.s(),  add.s(4) | add.s(8)) | xsum.s())
Out[4]: 20

# From here, things can go back to a normal chain
In [5]: ~(add.si(2, 2) | group(noop.s(),  add.s(4) | add.s(8)) | xsum.s() | add.s(1) | add.s(1))
Out[5]: 22

希望有用！

Celery：访问链中的所有先前结果

Celery: access all previous results in a chain

python

redis

celery