使用 Kubeflow Pipelines kfp.ParallelFor 时聚合结果答案

【问题标题】：Aggregate results when using Kubeflow Pipelines kfp.ParallelFor使用 Kubeflow Pipelines kfp.ParallelFor 时聚合结果
【发布时间】：2020-07-01 00:23:51
【问题描述】：

从 Kubeflow Pipleine kfp.ParallelFor 聚合结果的好模式是什么？

【问题讨论】：

标签： kubeflow kubeflow-pipelines

【解决方案1】：

目前可能不支持：

Support inputs with multiple arguments #1933

【讨论】：

【解决方案2】：

不完全是您要求的，但我们的解决方法是将 parallelfor 任务的结果写入 S3，然后在后处理任务中简单地收集它们。

with dsl.ParallelFor(preprocessing_task.output) as plant_item:
                predict_plant='{}'.format(plant_item)
                forecasting_task = forecasting_op(predict_plant, ....).after(preprocessing_task)
postprocessing_task = postprocessing_op(...).after(forecasting_task)

【讨论】：

（经过多次建议的编辑：不，后处理步骤不是在循环内，而是在循环之后。这正是收集结果的原因。）
您是否知道有任何文档支持这一点？当我尝试重新创建这种方法时，一个 postprocessing_task 节点出现在我的每个 forecasting_task 等效项的图表中。
您是否将 plant_item 或 predict_plant 作为输入？因为这样就可以理解了。但是不，我找不到文档，只是反复试验。