AI 数据处理管道：Argo Workflows 加 TKE 构建高效 ETL 流水线

# 创建命名空间
kubectl create namespace argo

# 通过 Helm 安装 Argo Workflows
helm install argo-workflows argo/argo-workflows \
  --namespace argo \
  --set server.authMode=server \
  --set server.extraArgs="{--auth-mode=server}"

kubectl -n argo port-forward svc/argo-workflows-server 2746:2746

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: data-pipeline-
  namespace: default
spec:
  entrypoint: data-pipeline
  arguments:
    parameters:
      - name: dataset-url
        value: "https://example.com/data/raw.tar.gz"
      - name: output-path
        value: "/data/processed"
  templates:
    - name: data-pipeline
      dag:
        tasks:
          - name: download
            template: download-data
            arguments:
              parameters:
                - name: url
                  value: "{{workflow.parameters.dataset-url}}"
          - name: convert
            template: convert-format
            dependencies: [download]
            arguments:
              parameters:
                - name: input-path
                  value: "{{tasks.download.outputs.parameters.output-path}}"
          - name: validate
            template: validate-quality
            dependencies: [convert]

    - name: download-data
      inputs:
        parameters:
          - name: url
      outputs:
        parameters:
          - name: output-path
            valueFrom:
              path: /tmp/output-path.txt
      container:
        image: python:3.11-slim
        command: [sh, -c]
        args:
          - |
            mkdir -p /data/raw && cd /data/raw && \
            wget -O data.tar.gz "{{inputs.parameters.url}}" && \
            tar -xzf data.tar.gz && \
            echo "/data/raw" > /tmp/output-path.txt
        volumeMounts:
          - name: data-volume
            mountPath: /data
      resources:
        requests:
          cpu: "1"
          memory: 2Gi

    - name: convert-format
      inputs:
        parameters:
          - name: input-path
      container:
        image: python:3.11-slim
        command: [python, -c]
        args:
          - |
            import os, json
            # 数据格式转换逻辑
            input_dir = "{{inputs.parameters.input-path}}"
            output_dir = "/data/converted"
            os.makedirs(output_dir, exist_ok=True)
            for f in os.listdir(input_dir):
                if f.endswith('.csv'):
                    # 转换为 Parquet 格式
                    pass
        volumeMounts:
          - name: data-volume
            mountPath: /data
      resources:
        limits:
          nvidia.com/gpu: "1"

    - name: validate-quality
      container:
        image: python:3.11-slim
        command: [python, -c]
        args:
          - |
            # 数据质量校验逻辑
            # 检查缺失值、异常值、数据分布等
            pass
        volumeMounts:
          - name: data-volume
            mountPath: /data

  volumeClaimTemplates:
    - metadata:
        name: data-volume
      spec:
        accessModes: ["ReadWriteOnce"]
        resources:
          requests:
            storage: 100Gi

kubectl create -f data-pipeline.yaml

apiVersion: v1
kind: PersistentVolume
metadata:
  name: pv-cfs-turbo-data
spec:
  accessModes:
    - ReadWriteMany
  capacity:
    storage: 1Ti
  csi:
    driver: com.tencent.cloud.csi.cfsturbo
    volumeHandle: pv-cfs-turbo-data
    volumeAttributes:
      proto: lustre
      rootdir: /cfs
      fsid: "<FSID>"
      host: "<MOUNT_IP>"
      path: /ai-data
  storageClassName: ""
---
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: pvc-cfs-turbo-data
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 1Ti
  volumeName: pv-cfs-turbo-data
  storageClassName: ""

volumes:
  - name: data-volume
    persistentVolumeClaim:
      claimName: pvc-cfs-turbo-data

apiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: daily-data-pipeline
  namespace: default
spec:
  schedule: "0 2 * * *"  # 每天凌晨 2 点执行
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 1
  workflowSpec:
    workflowTemplateRef:
      name: data-pipeline-template
    arguments:
      parameters:
        - name: dataset-url
          value: "https://example.com/data/daily-latest.tar.gz"

# 查看所有工作流
kubectl get workflows

# 查看特定工作流的详细状态
kubectl get workflow data-pipeline-xxxx -o yaml

# 实时查看日志
kubectl logs -f workflow-workflowname-step-name