批处理系统：Batch批量计算与云原生Serverless Argo Workflows-阿里云开发者社区

【阅读原文】戳：批处理系统：Batch批量计算与云原生Serverless Argo Workflows

随着自动驾驶、科学计算等领域对技术需求的不断深化，以及Kubernetes生态系统日趋丰富，容器化已成为批处理任务执行的主流模式。面对这一趋势，市场提供了两大类解决方案：一类是以Batch批量计算为代表的云服务商自主研发的封闭式平台，另一类则是围绕开源项目Argo Workflows构建的开放兼容平台。

对于企业研发团队而言，明智地选择符合自身业务需求的批处理任务平台至关重要，这直接关系到开发效率、成本控制及未来技术的可扩展性。本文将以一个典型的数据处理应用场景为案例，深入对比Batch批量计算与Argo Workflows的核心特性及适用场景，辅助技术决策者做出更加贴切的选择。

一、案例

如下图所示是一个典型的数据处理任务，第一步使用64个Pod进行数据处理，将128个文件合并成成64个文件，第二步使用32个Pod进行第二步的数据处理，将64个文件合并成32个文件，最后一步启动一个Pod进行最终结果的计算并输出到对象存储中。

架构图：

二、通过Batch批量计算实现

1. 原理

Batch批量计算通常是一项完全托管服务，可让您以任何规模运行批处理式工作负载。以下流程描述 Batch批量计算如何运行每个作业。

1. 创建作业定义，其指定如何运行作业，同时提供权限、内存和CPU要求以及其他配置选项。

2. 将作业提交到托管的Batch批量计算作业队列，作业将一直驻留在该队列中，直到被安排在计算环境中进行处理。

3. Batch批量计算计算队列中每个作业的CPU、内存和GPU要求，并在计算环境中调度计算资源以处理作业。

4. Batch批量计算调度程序将作业放入相应的Batch计算环境进行处理。

5. 作业成功或失败退出，将输出结果写入用户定义的存储空间。

2. 创建任务定义

第一步任务定义，构建process-data、merge-data等任务定义，需要准备镜像，启动参数，所需资源等。Batch批量计算服务通常提供友好的控制台交互，为易于编程和避免频繁的控制台演示，我们直接使用Json的方式定义任务。

a. process-data

{
  "type": "container",
  "containerProperties": { # 执行命令
    "command": [
      "python",
      "process.py"
    ],
    "image": "python:3.11-amd", # 镜像地址
    "resourceRequirements": [ # 资源需要
      {
        "type": "VCPU", 
        "value": "1.0"
      },
      {
        "type": "MEMORY",
        "value": "2048"
      }
    ],
    "runtimePlatform": {
      "cpuArchitecture": "X86_64", # cpu架构
      "operatingSystemFamily": "LINUX"
    },
    "networkConfiguration": {
      "assignPublicIp": "DISABLED"
    },
    "executionRoleArn": "role::xxxxxxx", # 权限
  },
  "platformCapabilities": [ # 后端资源：服务器或者Serverless Container
    "Serverless Container"
  ],
  "jobDefinitionName": "process-data" # Job名称
}

b. merge-data:

{
  "type": "container",
  "containerProperties": { # 执行命令，merge.py
    "command": [
      "python",
      "merge.py"
    ],
    "image": "python:3.11-amd", # 镜像
    "resourceRequirements": [ # 资源需要
      {
        "type": "VCPU",
        "value": "1.0"
      },
      {
        "type": "MEMORY",
        "value": "2048"
      }
    ],
    "runtimePlatform": {
      "cpuArchitecture": "X86_64",
      "operatingSystemFamily": "LINUX"
    },
    "networkConfiguration": {
      "assignPublicIp": "ENABLED"
    },
    "executionRoleArn": "role::xxxx", # 权限
    "repositoryCredentials": {},
  },
  "platformCapabilities": [ # 后端资源：服务器或者Serverless Container
    "Serverless Container"
  ],
  "jobDefinitionName": "merge-data" # Job名称
}

3. 提交任务并构建依赖关系

a. 定义并提交process-data-l1 Job

Job定义：

{
  "jobName": "process-data-l1",
  "jobDefinition": "arn::xxxx:job-definition/process-data:1", # Job使用的定义
  "jobQueue": "arn::xxxx:job-queue/process-data-queue", # Job使用的队列
  "dependsOn": [],
  "arrayProperties": { # 启动任务数
    "size": 64
  },
  "retryStrategy": {},
  "timeout": {},
  "parameters": {},
  "containerOverrides": {
    "resourceRequirements": [],
    "environment": []
  }
}

提交获取Job id：

# batch submit process-data-l1 | get job-id
job-id: b617f1a3-6eeb-4118-8142-1f855053b347

b. 提交process-data-l2 Job

该Job依赖process-data-l1 Job。

{
  "jobName": "process-data-l2",
  "jobDefinition": "arn::xxxx:job-definition/process-data:2", # Job使用的定义
  "jobQueue": "arn::xxxx:job-queue/process-data-queue", # Job使用的队列
  "dependsOn": [
    {
      "jobId": "b617f1a3-6eeb-4118-8142-1f855053b347" # process-data-l1的job Id
    }
  ],
  "arrayProperties": { # 启动任务数
    "size": 32
  },
  "retryStrategy": {},
  "timeout": {},
  "parameters": {},
  "containerOverrides": {
    "resourceRequirements": [],
    "environment": []
  }
}

提交获取Job id：

# batch submit process-data-l2 | get job-id
job-id: 6df68b3e-4962-4e4f-a71a-189be25b189c

c. 提交merge-data Job

该Job依赖process-data-l2 Job。

{
  "jobName": "merge-data",
  "jobDefinition": "arn::xxxx:job-definition/merge-data:1", # Job使用的定义
  "jobQueue": "arn::xxxx:job-queue/process-data-queue", # Merge Job使用的队列
  "dependsOn": [
    {
      "jobId": "6df68b3e-4962-4e4f-a71a-189be25b189c" # process-data-l2的job Id
    }
  ],
  "arrayProperties": {},
  "retryStrategy": {},
  "timeout": {},
  "parameters": {},
  "containerOverrides": {
    "resourceRequirements": [],
    "environment": []
  }
}

提交Job：

batch submit merge-data

4. 观察任务运行

所有任务均按序正常运行。

三、通过Argo Workflows实现

1. 原理

Serverless Argo Workflows是阿里云的一项全托管服务，基于开源Argo Workflow项目构建，完全符合开源工作流标准，可以让您在Kubernetes上运行任何规模的批处理负载，采用无服务器模式，使用阿里云弹性容器实例ECI运行工作流，通过优化Kubernetes集群参数，实现大规模工作流的高效弹性调度，同时配合抢占式ECI实例，优化成本。下面流程介绍Serverless Argo如何运行每个作业:

1）作业定义，指定如何运行作业，包括每个作业的CPU、内存、镜像、执行命令等。作业依赖关系定义，包括串行、并行循环、重试等。

2）提交Workflow到Serverless Argo集群。

3）Serverless Argo评估每个作业的资源，并调度弹性实例运行作业。

2. 创建任务定义并构建依赖关系

在构建Workflow时，定义Job和其依赖关系在同一个文件中定义。

第一步，定义process-data、和merge-data任务Template，描述每个任务的的镜像和启动参数。

第二步，定义Step/DAG Template，描述任务的并、串行执行关系。

第三步，将Template、依赖关系，存储、输入参数等整合成Workflow。

构建Workflow通常有两种方式，一种是通过Yaml方式构建，另一种是通过Python SDK构建，下边分别展示这两种方式的构建方法。

a. 通过Yaml方式构建

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: process-data- # 数据处理工作流
spec:
  entrypoint: main
  volumes: # 对象存储挂载
    - name: workdir
      persistentVolumeClaim:
        claimName: pvc-oss
  arguments:
    parameters:
      - name: numbers
        value: "64"
  templates:
    - name: main
      steps:
        - - name: process-data-l1 # 第一级处理，启动64个Pods，Merge 128 个files
            template: process-data
            arguments:
              parameters:
                - name: file_number
                  value: "{{item}}"
                - name: level
                  value: "1"
            withSequence:
              count: "{{workflow.parameters.numbers}}"
        - - name: process-data-l2 # 第二级处理，启动32个Pods，Merge 64 个files， 上一步处理完后启动
            template: process-data
            arguments:
              parameters:
                - name: file_number
                  value: "{{item}}"
                - name: level
                  value: "2"
            withSequence:
              count: "{{=asInt(workflow.parameters.numbers)/2}}"
        - - name: merge-data # 最后一级处理，启动一个Pod，Merge 32 files， 上一步处理完后启动
            template: merge-data
            arguments:
              parameters:
                - name: number
                  value: "{{=asInt(workflow.parameters.numbers)/2}}"
    - name: process-data # process-data 任务定义
      inputs:
        parameters:
          - name: file_number
          - name: level
      container:
        image: argo-workflows-registry.cn-hangzhou.cr.aliyuncs.com/argo-workflows-demo/python:3.11-amd
        imagePullPolicy: Always
        command: [python3] # command
        args: ["process.py", "{{inputs.parameters.file_number}}", "{{inputs.parameters.level}}"]# 接收输入的参数，启动多少个pod进行处理
        volumeMounts:
        - name: workdir
          mountPath: /mnt/vol
    - name: merge-data # merge-data 任务定义
      inputs:
        parameters:
          - name: number
      container:
        image: argo-workflows-registry.cn-hangzhou.cr.aliyuncs.com/argo-workflows-demo/python:3.11-amd
        imagePullPolicy: Always
        command: [python3]
        args: ["merge.py", "0", "{{inputs.parameters.number}}"] # 接收输入的参数，处理多少个文件。
        volumeMounts:
        - name: workdir
          mountPath: /mnt/vol

提交Workflow:

argo submit process-data.yaml

b. 通过Python SDK构建

from hera.workflows import Container, Parameter, Steps, Workflow, Volume
import urllib3
urllib3.disable_warnings()
# 配置访问地址和token
global_config.host = "https://argo.{{clusterid}}.{{region-id}}.alicontainer.com:2746"
global_config.token = "abcdefgxxxxxx" # 填入集群的token
global_config.verify_ssl = ""
with Workflow(
    generate_name="process-data-", # 数据处理工作流
    entrypoint="main",
    volumes=[m.Volume(name="workdir", persistent_volume_claim={"claim_name": "pvc-oss"})], # 对象存储挂载
    arguments=[Parameter(name="numbers", value="64")]
) as w:
    process-data = Container( # process-data 任务定义
        name="process-data",
        inputs=[Parameter(name="file_number"), Parameter(name="level")],
        image="argo-workflows-registry.cn-hangzhou.cr.aliyuncs.com/argo-workflows-demo/python:3.11-amd",
        command=["python3"],
        args=["process.py","{{inputs.parameters.file_number}}", "{{inputs.parameters.level}}"],
        volume_mounts=[
            m.VolumeMount(name="workdir", mount_path="/mnt/vol"),
        ],
    )
    merge-data = Container( # merge-data 任务定义
        name="merge-data",
        inputs=[Parameter(name="number")],
        image="argo-workflows-registry.cn-hangzhou.cr.aliyuncs.com/argo-workflows-demo/python:3.11-amd",
        command=["python3"],
        args=["merge.py", "0", "{{inputs.parameters.number}}"],
        volume_mounts=[
            m.VolumeMount(name="workdir", mount_path="/mnt/vol"),
        ],
    )
    with Steps(name="main") as s:
        process-data( 
            name="process-data-l1",
            arguments=[Parameter(name="file_number", value="{{item}}"), Parameter(name="level", value="1")],
        ) # 第一级处理，启动64个Pods，Merge 128 个files
        process-data(
            name="process-data-l2",
            arguments=[Parameter(name="file_number", value="{{item}}"), Parameter(name="level", value="2")],
        ) # 第二级处理，启动32个Pods，Merge 64 个files， 上一步处理完后启动
        merge-data(
            name="merge-data",
            arguments=[Parameter(name="number", value="{{=asInt(workflow.parameters.numbers)/2}}")],
        ) # 最后一级处理，启动一个Pod，Merge 32 files， 上一步处理完后启动
# 创建workflow
w.create()

提交任务：

python process.py

3. 提交并观察任务运行

通过Yaml或者Python SDK方式构建并提交后，即可在Argo Server控制台查看工作流运行状态。

工作流按序正常执行成功。

四、对比

可以看出Serverless Argo Workflows和Batch批量计算对容器批处理都有非常完善的支持。尽管它们的核心目标相似，但在任务定义、使用场景、灵活性以及资源管理等方面存在一些关键差异。以下是一个简要的对比。

五、总结

Serverless Argo Workflows和Batch批量计算对容器批处理都有非常完善的支持，选择Argo Workflows还是Batch批量计算主要取决于您的的技术栈、对云供应商的依赖程度、工作流的复杂性和对控制权的需求。如果您的团队熟悉Kubernetes并且需要高度定制化的工作流，Argo Workflows可能是更好的选择。相反，如果你在云厂商生态系统内运作，寻求简单易用且能与云厂商其他服务紧密集成的解决方案，Batch批量计算可能更适合您。

ACK One Serverles Argo团队是国内最早使用和维护Argo Workflows的团队之一，在Argo Workflows使用方面积累众多的最佳实践，如果您需要使用Argo Workflows调度大规模工作流，欢迎加入钉钉群号一同交流：35688562。

参考：

全托管Serverless Argo工作流：

https://www.alibabacloud.com/help/zh/ack/distributed-cloud-container-platform-for-kubernetes/user-guide/overview-12

Argo Workflows：

https://github.com/argoproj/argo-workflows

创建工作流集群：

https://www.alibabacloud.com/help/zh/ack/distributed-cloud-container-platform-for-kubernetes/user-guide/create-a-workflow-cluster

PythonSDK助力大规模Argo Workflows构建：

https://mp.weixin.qq.com/s/_2Glhuy6YJEM4ZRMDwE7JA

我们是阿里巴巴云计算和大数据技术幕后的核心技术输出者。

欢迎关注 “阿里云基础设施”同名微信、微博、知乎