如何在Argo中针对大输出结果并行运行任务？

当你需要对十万级别的列表条目逐个执行任务时，原模板直接用withParam引用前一步的输出结果会触发两个核心问题：一是Argo的参数（parameters）有大小限制，无法承载超大体积的JSON列表；二是一次性创建十万个任务会导致Kubernetes API服务器过载、资源耗尽，最终流程崩溃。结合Artifact存储和分批次处理可以解决这个问题，具体方案如下：

核心思路 #

1. 用Artifact替代参数存储大列表：Artifact支持存储大文件，不受参数大小限制，适合保存十万级别的条目列表。
2. 分批次迭代处理：将大列表拆分为多个小批次，控制并发任务数量，避免一次性压垮集群。

修改后的完整Workflow模板 #

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: example-large-output-
spec:
  entrypoint: main
  templates:
    - name: main
      steps:
        # 第一步：生成超大列表并以Artifact输出
        - - name: create-large-output
            template: create-large-output-template
        # 第二步：将大列表拆分为小批次
        - - name: split-list-into-batches
            template: split-large-list
            arguments:
              artifacts:
                - name: large-list
                  from: "{{steps.create-large-output.outputs.artifacts.large-list}}"
        # 第三步：迭代处理每个批次
        - - name: process-batch
            template: process-batch-template
            arguments:
              parameters:
                - name: batch
                  value: "{{item}}"
            withParam: "{{steps.split-list-into-batches.outputs.result}}"

    # 生成超大列表，输出为Artifact
    - name: create-large-output-template
      script:
        image: python:3.9-slim
        command: [python]
        source: |
          import json
          import sys
          # 定义单个条目内容
          item_content = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
          # 生成10万条目的列表
          large_list = [item_content for _ in range(100000)]
          # 将列表写入stdout，自动被Argo保存为Artifact
          json.dump(large_list, sys.stdout)
      outputs:
        artifacts:
          - name: large-list
            path: /tmp/script-output
            archive:
              none: {}  # 不压缩，直接存储

    # 将大列表拆分为小批次
    - name: split-large-list
      inputs:
        artifacts:
          - name: large-list
            path: /tmp/large-list.json
      script:
        image: python:3.9-slim
        command: [python]
        source: |
          import json
          import sys

          # 读取Artifact中的大列表
          with open('/tmp/large-list.json', 'r') as f:
              items = json.load(f)

          # 定义批次大小，可根据集群资源调整
          batch_size = 100
          # 拆分为多个小批次
          batches = [items[i:i+batch_size] for i in range(0, len(items), batch_size)]
          # 输出批次列表供后续迭代
          json.dump(batches, sys.stdout)

    # 处理单个批次，迭代批次内的每个条目
    - name: process-batch-template
      inputs:
        parameters:
          - name: batch
      steps:
        - - name: iterate-item
            template: iterate-large-output-template
            arguments:
              parameters:
                - name: fp
                  value: "{{item}}"
            withParam: "{{inputs.parameters.batch}}"
            parallelism: 5  # 限制每个批次的并发任务数，可调整

    # 单个条目的处理逻辑
    - name: iterate-large-output-template
      inputs:
        parameters:
          - name: fp
      script:
        image: alpine
        command:
          - sh
        source: |
          echo {{inputs.parameters.fp}}

关键优化点说明 #

1. Artifact存储大列表：将原本通过echo输出为参数的列表，改为用Python生成并通过Artifact输出，彻底规避参数大小限制。
2. 分批次拆分：通过Python脚本将十万条目的列表拆分为1000个批次（每个批次100条），每次仅处理一个批次的任务，大幅降低集群瞬时负载。
3. 并发控制：在批次处理步骤中添加parallelism字段，限制每个批次内同时运行的任务数量，进一步避免资源耗尽。

额外优化建议 #

• 调整批次大小：根据你的Kubernetes集群节点数、CPU/内存资源，修改batch_size的值，平衡处理速度和集群压力。
• 高效生成列表：用Python生成列表比bash循环效率高得多，尤其在生成十万级条目时，能节省大量时间。
• 资源限制：给每个处理任务的Pod添加resources字段，限制CPU和内存使用，避免单个Pod占用过多资源：

  - name: iterate-large-output-template
    inputs:
      parameters:
        - name: fp
    script:
      image: alpine
      command: [sh]
      source: |
        echo {{inputs.parameters.fp}}
      resources:
        limits:
          cpu: "100m"
          memory: "64Mi"
        requests:
          cpu: "50m"
          memory: "32Mi"
  

内容的提问来源于stack exchange，提问作者olli