使用Argo Workflow编排动态DAG Fan-out/Fan-in任务_容器服务 Kubernetes 版 ACK(ACK)-阿里云帮助中心

在工作流编排过程中，为了加快大任务处理的速度，可以使用Fan-out Fan-in任务编排，将大任务分解成小任务，然后并行运行小任务，最后聚合结果。分布式工作流Argo集群（简称工作流集群）支持动态DAG方式编排Fan-out Fan-in任务，可按需调度云上算力、利用云上弹性可调用数万核CPU资源，减少运行时间，运行结束后能够及时回收资源节省成本。本文为您介绍如何使用工作流集群的Argo Workflow编排动态DAG Fan-out Fan-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）。

操作流程

创建分布式工作流Argo集群。
挂载阿里云OSS存储卷，以便工作流可以像操作本地文件一样操作OSS上的文件。
具体操作，请参见使用存储卷。

使用以下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

使用动态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。
工作流运行后，您可以在ACK One工作流集群控制台查看任务DAG流程与运行结果。
阿里云OSS文件列表中，log-count-data.txt为输入日志文件，split-output，cout-output为中间结果目录，result.json为最终结果文件。