如何递归链接将一个列表归为一组的Celery任务?

问题描述 投票:1回答:1

我从这个问题开始:How to chain a Celery task that returns a list into a group?

但是我想扩大两次。所以在我的用例中,我有:

  • 任务A:确定给定日期的项目总数
  • 任务B:在该日期下载1000个元数据条目
  • 任务C:下载一项内容

因此,每一步我都在扩展下一步的项目数量。我可以通过遍历任务中的结果并在下一个任务函数上调用.delay()来实现。但是我想我尽量不要让我的主要任务那样做。相反,他们将返回一个元组列表-然后,每个元组都将扩展为用于调用下一个函数的参数。

上述问题的答案似乎可以满足我的需要,但是我无法为两级扩展制定出正确的链接方法。

这是我的代码的非常精简的示例:

from celery import group
from celery.task import subtask
from celery.utils.log import get_task_logger

from .celery import app

logger = get_task_logger(__name__)

@app.task
def task_range(upper=10):
    # wrap in list to make JSON serializer work
    return list(zip(range(upper), range(upper)))

@app.task
def add(x, y):
    logger.info(f'x is {x} and y is {y}')
    char = chr(ord('a') + x)
    char2 = chr(ord('a') + x*2)
    result = x + y
    logger.info(f'result is {result}')
    return list(zip(char * result, char2 * result))

@app.task
def combine_log(c1, c2):
    logger.info(f'combine log is {c1}{c2}')

@app.task
def dmap(args_iter, celery_task):
    """
    Takes an iterator of argument tuples and queues them up for celery to run with the function.
    """
    logger.info(f'in dmap, len iter: {len(args_iter)}')
    callback = subtask(celery_task)
    run_in_parallel = group(callback.clone(args) for args in args_iter)
    return run_in_parallel.delay()

然后,我尝试了各种方法来使嵌套映射起作用。首先,单级映射可以正常工作,因此:

pp = (task_range.s() | dmap.s(add.s()))
pp(2)

产生我期望的结果,所以我并不完全满意。

但是当我尝试添加另一个级别时:

ppp = (task_range.s() | dmap.s(add.s() | dmap.s(combine_log.s())))

然后在工作程序中看到错误:

[2019-11-23 22:34:12,024: ERROR/ForkPoolWorker-2] Task proj.tasks.dmap[e92877a9-85ce-4f16-88e3-d6889bc27867] raised unexpected: TypeError("add() missing 2 required positional arguments: 'x' and 'y'",)
Traceback (most recent call last):
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/app/trace.py", line 385, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/app/trace.py", line 648, in __protected_call__
    return self.run(*args, **kwargs)
  File "/home/hdowner/dev/playground/celery/proj/tasks.py", line 44, in dmap
    return run_in_parallel.delay()
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/canvas.py", line 186, in delay
    return self.apply_async(partial_args, partial_kwargs)
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/canvas.py", line 1008, in apply_async
    args=args, kwargs=kwargs, **options))
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/canvas.py", line 1092, in _apply_tasks
    **options)
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/canvas.py", line 578, in apply_async
    dict(self.options, **options) if options else self.options))
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/canvas.py", line 607, in run
    first_task.apply_async(**options)
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/canvas.py", line 229, in apply_async
    return _apply(args, kwargs, **options)
  File "/home/hdowner/.venv/play_celery/lib/python3.6/site-packages/celery/app/task.py", line 532, in apply_async
    check_arguments(*(args or ()), **(kwargs or {}))
TypeError: add() missing 2 required positional arguments: 'x' and 'y'

而且我不确定为什么将dmap()的参数从普通任务签名更改为链条会改变将参数传递到add()的方式。我的印象是不应该这样,它只意味着将传递add()的返回值。但显然并非如此...

python celery
1个回答
0
投票

结果是,clone()实例上的chain方法在某些时候没有传递参数-有关完整详细信息,请参见https://stackoverflow.com/a/53442344/3189。如果我在该答案中使用该方法,则我的dmap()代码变为:

@app.task
def dmap(args_iter, celery_task):
    """
    Takes an iterator of argument tuples and queues them up for celery to run with the function.
    """
    callback = subtask(celery_task)
    run_in_parallel = group(clone_signature(callback, args) for args in args_iter)
    return run_in_parallel.delay()


def clone_signature(sig, args=(), kwargs=(), **opts):
    """
    Turns out that a chain clone() does not copy the arguments properly - this
    clone does.
    From: https://stackoverflow.com/a/53442344/3189
    """
    if sig.subtask_type and sig.subtask_type != "chain":
        raise NotImplementedError(
            "Cloning only supported for Tasks and chains, not {}".format(sig.subtask_type)
        )
    clone = sig.clone()
    if hasattr(clone, "tasks"):
        task_to_apply_args_to = clone.tasks[0]
    else:
        task_to_apply_args_to = clone
    args, kwargs, opts = task_to_apply_args_to._merge(args=args, kwargs=kwargs, options=opts)
    task_to_apply_args_to.update(args=args, kwargs=kwargs, options=deepcopy(opts))
    return clone

然后当我这样做时:

ppp = (task_range.s() | dmap.s(add.s() | dmap.s(combine_log.s())))

一切正常。

© www.soinside.com 2019 - 2024. All rights reserved.