Сельдерей Создание групповых задач из цепочки задач

Question

Я пытаюсь связать следующие задачи с сельдереем (v4.0),

task = group([tasks1.s(), task2.s()) | generate_job_requests.s() | execute_job.map() | aggregate_result.s()
result = task.get()

Выше часть работает нормально до generate_job_requests как аккорд Но проблема начинается с execute_job где он получает список рабочих мест generate_job_requests, для которого мне нужно создать параллельные задачи, а затем совокупный результат всех заданий.

Я пытаюсь проверить, возможен ли такой вид графа с сельдереем? Есть ли возможный альтернативный рабочий процесс для решения проблемы с такой зависимостью? Все, что мне не хватает в документации.

Answer 1

Я использовал карту как функциональность с промежуточным создателем задачи, который действует как аккорд,

@shared_task(ignore_result=False)
def dmap(it, callback, end_task):
    callback = subtask(callback)
    grp = group(callback.clone([arg, ]) for arg in it)
    c = (grp | end_task)
    return c()

Таким образом, поток задач был сокращен, как это,

task = (group([tasks1.s(), task2.s()) | generate_job_requests.s() | dmap.s(
        execute_job.s(), aggregate_result.s())).apply_async()

Для получения окончательного результата задачи я сделал несколько твиков,

# As we get dmap task id here
dmap_task = celery_app.AsyncResult(task.id)
dmap_result = dmap_task.get()
# Get actual aggregate_result task id
aggr_res_task_id = dmap_result[0][0]
result = celery_app.AsyncResult(aggr_res_task_id)
# Here we receive actual output of overall task
result.get()

Я передал ответ