Argo Workflows K8s 原生工作流

bash kubectl create namespace argo kubectl apply -n argo -f https://github.com/argoproj/argo-workflows/releases/latest/download/install.yaml

如果你的集群版本较新或对安全有限制，请查阅官方文档选择合适的安装清单。

#### 2. 访问 Argo UI

安装完成后，通过端口转发访问 Web 界面：

```bash
kubectl -n argo port-forward deployment/argo-server 2746:2746

浏览器打开 https://localhost:2746，首次访问会提示安全问题，因为默认使用自签名证书，点击“高级”继续即可。你可以通过 Argo Server 提供的 API 或 UI 提交和管理工作流。

3. 运行你的第一个工作流

创建一个简单的 Hello World 工作流 YAML 文件 hello-workflow.yaml：

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: hello-world-
spec:
  entrypoint: whalesay
  templates:
  - name: whalesay
    container:
      image: docker/whalesay
      command: [cowsay]
      args: ["Hello Argo!"]

提交工作流：

kubectl -n argo create -f hello-workflow.yaml

使用 Argo CLI（需单独安装）或 UI 查看运行状态。也可以直接用 kubectl 查看该 Pod 的日志。

核心概念深度解析

工作流（Workflow）

一个 Workflow 对象是 Argo 中的顶级资源，定义了一次执行。它由 entrypoint 指定第一个执行的模板，并包含一个或多个模板。

模板（Template）

模板是可重用的任务定义，可以是以下几种类型：

• 容器模板（Container）：运行一个容器，最基础的形式。
• 脚本模板（Script）：在容器内运行一段脚本，自动注入脚本内容。
• 资源模板（Resource）：直接操作 Kubernetes 资源（创建/获取/删除）。
• 挂起模板（Suspend）：暂停工作流，等待手动恢复或超时。
• 步骤模板（Steps）：将多个模板按步骤组织，支持并行执行。
• DAG模板（DAG）：以有向无环图的方式定义任务之间的依赖。

步骤（Steps）与 DAG

步骤模板使用列表形式定义一组串行或并行的步骤：

templates:
- name: my-steps
  steps:
  - - name: step-A
      template: task-one
    - name: step-B
      template: task-two
  - - name: step-C
      template: task-three

上面数组的每个元素代表一个步骤组，同一组内的步骤并行执行，不同组之间串行执行。例如 step-A 和 step-B 并行，然后 step-C 在它们完成后执行。

DAG 模板通过 dependencies 显式指定依赖：

templates:
- name: my-dag
  dag:
    tasks:
    - name: A
      template: task
    - name: B
      template: task
      dependencies: [A]
    - name: C
      template: task
      dependencies: [A]
    - name: D
      template: task
      dependencies: [B, C]

参数与工件（Parameters & Artifacts）

• 参数：用于在模板之间传递简单值（字符串、数字等）。使用 {{inputs.parameters.xxx}} 和 {{outputs.parameters.xxx}} 声明和引用。
• 工件：用于传递文件或大型二进制数据，支持 S3、GCS、Azure Blob 或 PVC 等存储。Argo 会自动处理压缩、打包和移动。

循环、条件与递归

• 循环：使用 withSequence（数字范围）、withItems（列表）或 withParam（动态参数）实现。
• 条件：通过 when 表达式支持条件执行，如 "{{inputs.parameters.action}} == 'deploy'"。
• 递归：允许模板在某个条件下调用自身，实现类似树的遍历。

实战：构建一个简单的数据处理管道

场景：从一个源 S3 桶下载数据，用两个不同的处理脚本并发分析，最后聚合结果并上传。

工作流定义草图：

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: data-pipeline-
spec:
  entrypoint: pipeline
  templates:
  - name: pipeline
    steps:
    - - name: download
        template: download-data
    - - name: process-A
        template: process
        arguments:
          parameters:
          - name: script
            value: "a.py"
      - name: process-B
        template: process
        arguments:
          parameters:
          - name: script
            value: "b.py"
    - - name: merge
        template: merge-results
  - name: download-data
    container:
      image: amazon/aws-cli
      command: [sh, -c]
      args: ["aws s3 cp s3://my-bucket/input.csv /data/input.csv"]
      volumeMounts:
      - name: workdir
        mountPath: /data
  - name: process
    inputs:
      parameters:
      - name: script
    container:
      image: python:3.9
      command: [sh, -c]
      args: ["python /scripts/{{inputs.parameters.script}} --input /data/input.csv --output /data/output_{{inputs.parameters.script}}.csv"]
  - name: merge-results
    container:
      image: python:3.9
      command: [sh, -c]
      args: ["python /scripts/merge.py --inputs /data/output_a.py.csv,/data/output_b.py.csv --output /data/final.csv"]
      volumeMounts:
      - name: workdir
        mountPath: /data
  volumes:
  - name: workdir
    emptyDir: {}