python - Celery:异步检索上一个任务的部分结果

Question

我想链接我的 ETL 工作流程，由此 Load 任务可以异步传输来自 ExtractTransform 任务的部分结果，而无需等待 ExtractTransform 完成。 celery 可以吗？

我正在考虑的两种方法:

方法 1

创建一个 ETLTask，其中 LoadTask(以某种方式)不断从 ETLTask 获取部分结果并将其出列(本质上分离生产者和消费者)。我无法从 AsyncResult 看出如果可能的话。听起来我只是想走独立的生产者和消费者的路线，我不知道如何在 Celery 中做到这一点。

class ExtractTransformTask(Task):

    def long_running_extract_transform(self):
        pass

    def run(self):
        return self.long_running_extract_transform()

class LoadTask(Task):

    def long_running_load(self):
        pass

    def run(self, results):
        self.long_running_load(results)

class ETLTask(Task):

    def run(self):
        et_result = ExtractTransformTask.delay()
        # while et_result PENDING or SUCCESS
        # dequeue current results and load with LoadTask instance

方法 2

分块提取源数据并创建多个加载任务。

Answer 1

使用方法 2 的解决方案。

class ExtractTransformMixin(object):

    def long_running_extract_transform(self, chunkify=False):
        pass

class LoadTask(Task):

    def long_running_load(self):
        pass

    def run(self, results):
        self.long_running_load(results)

class ETLTask(ExtractTransformMixin, Task):

    def run(self):
        load_results = ResultSet([])
        for chunk in long_running_extract_transform(chunkify=True):
            load_results.add(LoadTask().delay(chunk))
        return load_results