使用Argo Workflow编排动态DAG Fan-outFan-in任务

背景信息

Fan-out Fan-in

Fan-out和Fan-in常用于构建高效的并发处理流程，通过拆分（Fan-out）和聚合（Fan-in）操作，能够充分利用多核、多机资源，实现大规模数据的高效处理。

如上图所示，工作流编排过程中，可以使用DAG（有向无环图）编排Fan-out Fan-in任务。子任务的拆分方式分为有静态（静态DAG）和动态（动态DAG）。

• 静态DAG：拆分的子任务分类是固定的。例如：在数据收集场景中，同时收集数据库1和数据库2中的数据，最后聚合结果。

• 动态DAG：拆分的子任务分类是动态的，取决于前一个任务的输出结果。

  如上图所示，在数据处理场景中，任务A可以扫描待处理的数据集，为每个子数据集（例如：一个子目录）启动子任务Bn处理，当所有子任务Bn运行结束后，在子任务C中聚合结果。具体启动多少个子任务B取决于任务A的输出结果，根据实际的业务场景，可以在任务A中自定义子任务的拆分规则。

ACK One分布式工作流Argo集群

在实际业务场景中，为了提升大任务的执行效率，往往需要将一个大任务拆分成数千个子任务，为了保证这数千个子任务的同时运行，需要调度数万核的CPU资源。叠加多任务需要竞争资源，一般IDC离线任务难以满足需求。

例如：自动驾驶仿真任务，修改算法后进行回归测试，需要对所有驾驶场景进行仿真，每个驾驶场景为一个子任务运行，研发团队为了提高迭代速度，会要求所有子场景测试并行执行。

基于以上业务场景，您可以使用ACK One分布式工作流Argo集群编排工作流，工作流集群支持托管Argo Workflow，提供完善的售后技术支持，可通过动态DAG方式编排Fan-out Fan-in任务，支持弹调度云上算力，调度数万核CPU资源支撑大规模子任务的并行运行。任务运行结束时能够及时回收资源节省成本，一般可用于数据处理、机器学习、仿真计算、CI/CD等业务。

Argo Workflow是开源CNCF毕业项目，聚焦云原生领域下的工作流编排，使用Kubernetes CRD编排离线任务和DAG工作流，并使用Kubernetes Pod在集群中调度运行。更多信息，请参见Argo Workflow。

使用Argo Workflow编排Fan-out Fan-in任务

任务背景

本文以如下任务为例，为您介绍具体操作流程。

构建一个一个动态DAG Fan-out Fan-in工作流，读取阿里云OSS对象存储中的一个大日志文件，将其拆分为多个小文件（split），启动多个子任务分别计算每个小文件中的关键词数量（count），最后聚合结果（merge）。

操作流程

1. 创建分布式工作流Argo集群。

2. 挂载阿里云OSS存储卷，以便工作流可以像操作本地文件一样操作OSS上的文件。

   具体操作，请参见使用存储卷。

3. 使用以下YAML创建一个工作流。

   具体操作，请参见创建工作流。

   展开查看YAML示例

   apiVersion: argoproj.io/v1alpha1
   kind: Workflow
   metadata:
     generateName: dynamic-dag-map-reduce-
   spec:
     entrypoint: main
     # claim a OSS PVC, workflow can read/write file in OSS through PVC. 
     volumes:
       - name: workdir
         persistentVolumeClaim:
           claimName: pvc-oss
     # how many tasks to split, default is 5.
     arguments:
       parameters:
         - name: numParts
           value: "5"
     templates:
       - name: main
         # DAG definition.
         dag:
           tasks:
             # split log files to several small files, based on numParts.
             - name: split
               template: split
               arguments:
                 parameters:
                   - name: numParts
                     value: "{{workflow.parameters.numParts}}"
             # multiple map task to count words in each small file.
             - name: map
               template: map
               arguments:
                 parameters:
                   - name: partId
                     value: '{{item}}'
               depends: "split"
               # run as a loop, partId from split task json outputs.
               withParam: '{{tasks.split.outputs.result}}'
             - name: reduce
               template: reduce
               arguments:
                 parameters:
                   - name: numParts
                     value: "{{workflow.parameters.numParts}}"
               depends: "map"
       # The `split` task split the big log file to several small files. Each file has a unique ID (partId).
       # Finally, it dumps a list of partId to stdout as output parameters
       - name: split
         inputs:
           parameters:
             - name: numParts
         container:
           image: acr-multiple-clusters-registry.cn-hangzhou.cr.aliyuncs.com/ack-multiple-clusters/python-log-count
           command: [python]
           args: ["split.py"]
           env:
           - name: NUM_PARTS
             value: "{{inputs.parameters.numParts}}"
           volumeMounts:
           - name: workdir
             mountPath: /mnt/vol
       # One `map` per partID is started. Finds its own "part file" and processes it.
       - name: map
         inputs:
           parameters:
             - name: partId
         container:
           image: acr-multiple-clusters-registry.cn-hangzhou.cr.aliyuncs.com/ack-multiple-clusters/python-log-count
           command: [python]
           args: ["count.py"]
           env:
           - name: PART_ID
             value: "{{inputs.parameters.partId}}"
           volumeMounts:
           - name: workdir
             mountPath: /mnt/vol
       # The `reduce` task takes the "results directory" and returns a single result.
       - name: reduce
         inputs:
           parameters:
             - name: numParts
         container:
           image: acr-multiple-clusters-registry.cn-hangzhou.cr.aliyuncs.com/ack-multiple-clusters/python-log-count
           command: [python]
           args: ["merge.py"]
           env:
           - name: NUM_PARTS
             value: "{{inputs.parameters.numParts}}"
           volumeMounts:
           - name: workdir
             mountPath: /mnt/vol
         outputs:
           artifacts:
             - name: result
               path: /mnt/vol/result.json

4. 使用动态DAG方式实现Fan-out Fan-in编排任务。

   1. 大文件拆分为多个split子任务后，会在标准输出中输出一个JSON字符串，包含子任务要处理的partId，例如：

      ["0", "1", "2", "3", "4"]

   2. map任务使用withParam引用split任务的输出，并解析JSON字符串获得所有{{item}}，并使用每个{{item}}作为输入参数启动多个map任务。

                - name: map
                  template: map
                  arguments:
                    parameters:
                      - name: partId
                        value: '{{item}}'
                  depends: "split"
                  withParam: '{{tasks.split.outputs.result}}'

   更多定义方式，请参见开源Argo Workflow。

5. 工作流运行后，您可以在ACK One工作流集群控制台查看任务DAG流程与运行结果。

   

6. 阿里云OSS文件列表中，log-count-data.txt为输入日志文件，split-output，cout-output为中间结果目录，result.json为最终结果文件。

   

示例源代码参考

argo-workflow-examples

相关参考

• 了解阿里云分布式工作流Argo集群的详细功能，请参见ACK One概述。

• 了解Argo Workflow的具体内容，请参见开源Argo Workflow。

• 如果您对于ACK One有任何反馈或疑问，请加入钉群（钉群号：35688562）联系我们。