Use Argo Workflows for batch data processing-Container Service for Kubernetes(ACK)-阿里云帮助中心

This topic shows how to use Argo Workflows to hierarchically merge files in OSS. This is a common batch data processing technique used for tasks like high-precision map processing and animation rendering. Argo Workflows uses task orchestration to increase data processing concurrency and speed.

Prerequisites

The Argo Workflows component installed and Alibaba Cloud Argo CLI configured. For setup instructions, see Enable batch task orchestration.
You have created an OSS volume as described in Create an OSS volume.

Step 1: Prepare the data

Create a file named prepare-data.yaml with the following content.

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: aggregation-prepare-data-
  namespace: argo
spec:
  entrypoint: main
  volumes:
    - name: workdir
      persistentVolumeClaim:
        claimName: pvc-oss
  templates:
    - name: main
      dag:
        tasks:
          - name: prepare-data
            template: create-file
    - name: create-file
      script:
        image: mirrors-ssl.aliyuncs.com/python:alpine
        command:
          - python
        source: |
          import os
          import sys
          import random
          os.makedirs("/mnt/vol/aggregation-demo/l1/", exist_ok=True)
          for i in range(32):
            with open('/mnt/vol/aggregation-demo/l1/' + str(i) + '.txt', 'w') as conbine_file:              
              combined_content = random.choice('ABCDEFGHIJKLMNOPQRSTUVWXYZ')
              conbine_file.write(combined_content)
        volumeMounts:
        - name: workdir
          mountPath: /mnt/vol

Run the following command to submit the workflow.
```
argo submit prepare-data.yaml
```
After the workflow completes, it creates a directory named aggregation-demo/l1 in OSS. This directory contains 32 files, each with a single random character.

Step 2: Process the batch data

Create a file named process-data.yaml with the following content.

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: process-data- # Data processing workflow.
  namespace: argo
spec:
  entrypoint: main
  volumes: # Mounts the OSS volume.
    - name: workdir
      persistentVolumeClaim:
        claimName: pvc-oss
  arguments:
    parameters:
      - name: numbers
        value: "16"
  templates:
    - name: main
      steps:
        - - name: process-data-l1 # Level 1 processing: starts 16 pods to merge 32 files.
            template: process-data
            arguments:
              parameters:
                - name: file_number
                  value: "{{item}}"
                - name: level
                  value: "1"
            withSequence:
              count: "{{workflow.parameters.numbers}}"
        - - name: process-data-l2 # Level 2 processing: starts 8 pods to merge 16 files after the previous step completes.
            template: process-data
            arguments:
              parameters:
                - name: file_number
                  value: "{{item}}"
                - name: level
                  value: "2"
            withSequence:
              count: "{{=asInt(workflow.parameters.numbers)/2}}"
        - - name: merge-data # Final processing level: starts one pod to merge 8 files after the previous step completes.
            template: merge-data
            arguments:
              parameters:
                - name: number
                  value: "{{=asInt(workflow.parameters.numbers)/2}}"
    - name: process-data # Definition of the process-data task.
      inputs:
        parameters:
          - name: file_number
          - name: level
      container:
        image: serverlessargo-registry.cn-hangzhou.cr.aliyuncs.com/argo-workflow-examples/python:3.11-amd # Note: Make sure your Virtual Private Cloud (VPC) has internet access to pull the image.
        imagePullPolicy: Always
        command: [python3] # Command to run.
        args: ["process.py", "{{inputs.parameters.file_number}}", "{{inputs.parameters.level}}"]# Receives input parameters specifying which file the pod should process.
        volumeMounts:
        - name: workdir
          mountPath: /mnt/vol
    - name: merge-data # Definition of the merge-data task.
      inputs:
        parameters:
          - name: number
      container:
        image: serverlessargo-registry.cn-hangzhou.cr.aliyuncs.com/argo-workflow-examples/python:3.11-amd
        imagePullPolicy: Always
        command: [python3]
        args: ["merge.py", "{{inputs.parameters.number}}"] # Receives input parameters specifying the number of files to process.
        volumeMounts:
        - name: workdir
          mountPath: /mnt/vol

Run the following command to submit the workflow.
```
argo submit process-data.yaml -n argo
```

Run the following command to view the result.

argo get @latest -n argo

Expected output:

argo get @latest -n argo
Name:                process-data-8sn2q
Namespace:           argo
ServiceAccount:      unset (will run with the default ServiceAccount)
Status:              Succeeded
Conditions:          
 PodRunning          False
 Completed           True
Created:             Thu Dec 12 13:15:39 +0800 (4 minutes ago)
Started:             Thu Dec 12 13:15:39 +0800 (4 minutes ago)
Finished:            Thu Dec 12 13:16:40 +0800 (3 minutes ago)
Duration:            1 minute 1 seconds
Progress:            25/25
ResourcesDuration:   4s*(1 cpu),2m51s*(100Mi memory)
Parameters:          
  numbers:           16
STEP                           TEMPLATE      PODNAME                                     DURATION  MESSAGE
 ✔ process-data-8sn2q          main                                                                  
 ├─┬─✔ process-data-l1(0:0)    process-data  process-data-8sn2q-process-data-3064646785  17s         
 │ ├─✔ process-data-l1(1:1)    process-data  process-data-8sn2q-process-data-140728989   20s         
 │ ├─✔ process-data-l1(2:2)    process-data  process-data-8sn2q-process-data-499182361   18s         
 │ ├─✔ process-data-l1(3:3)    process-data  process-data-8sn2q-process-data-3152865965  20s         
 │ ├─✔ process-data-l1(4:4)    process-data  process-data-8sn2q-process-data-1363784105  16s         
 │ ├─✔ process-data-l1(5:5)    process-data  process-data-8sn2q-process-data-3270437485  20s         
 │ ├─✔ process-data-l1(6:6)    process-data  process-data-8sn2q-process-data-1788045361  16s         
 │ ├─✔ process-data-l1(7:7)    process-data  process-data-8sn2q-process-data-913839549   20s         
 │ ├─✔ process-data-l1(8:8)    process-data  process-data-8sn2q-process-data-1562179905  16s         
 │ ├─✔ process-data-l1(9:9)    process-data  process-data-8sn2q-process-data-573517021   20s         
 │ ├─✔ process-data-l1(10:10)  process-data  process-data-8sn2q-process-data-3769586203  16s         
 │ ├─✔ process-data-l1(11:11)  process-data  process-data-8sn2q-process-data-3700909073  20s         
 │ ├─✔ process-data-l1(12:12)  process-data  process-data-8sn2q-process-data-2818003295  19s         
 │ ├─✔ process-data-l1(13:13)  process-data  process-data-8sn2q-process-data-278901825   20s         
 │ ├─✔ process-data-l1(14:14)  process-data  process-data-8sn2q-process-data-3986961347  16s         
 │ └─✔ process-data-l1(15:15)  process-data  process-data-8sn2q-process-data-2905592609  19s         
 ├─┬─✔ process-data-l2(0:0)    process-data  process-data-8sn2q-process-data-2056515729  9s          
 │ ├─✔ process-data-l2(1:1)    process-data  process-data-8sn2q-process-data-2141620461  4s          
 │ ├─✔ process-data-l2(2:2)    process-data  process-data-8sn2q-process-data-352538601   9s          
 │ ├─✔ process-data-l2(3:3)    process-data  process-data-8sn2q-process-data-2144734909  6s          
 │ ├─✔ process-data-l2(4:4)    process-data  process-data-8sn2q-process-data-2290907961  8s          
 │ ├─✔ process-data-l2(5:5)    process-data  process-data-8sn2q-process-data-4197561341  5s          
 │ ├─✔ process-data-l2(6:6)    process-data  process-data-8sn2q-process-data-1620613249  9s          
 │ └─✔ process-data-l2(7:7)    process-data  process-data-8sn2q-process-data-3126908173  10s         
 └───✔ merge-data              merge-data    process-data-8sn2q-merge-data-1171626309    7s

View the result in the console.

In the Argo Workflows console, the directed acyclic graph (DAG) for the process-data workflow has three levels: the first level consists of 16 parallel process-data-l1 nodes, the second level has 8 parallel process-data-l2 nodes, and the final level is a single merge-data aggregation node. A green checkmark on a node indicates that it completed successfully.

This confirms that all tasks completed successfully. You can now check OSS and find the final result.txt file.

Contact us

If you have any product suggestions or questions, you can contact us by joining the DingTalk group (ID: 35688562).

上一篇: Build large-scale Argo Workflows with the Python SDK 下一篇: Build a CI pipeline with Argo Workflows

Prerequisites

Step 1: Prepare the data

Step 2: Process the batch data

Related documentation

Contact us