Container Service for Kubernetes:Building large-scale Argo Workflows with the Python SDK

Background

Argo Workflows is an open-source workflow management tool designed for Kubernetes environments, specializing in the automated orchestration of complex workflows. It allows you to define a series of tasks and flexibly arrange their execution order and dependencies. With Argo Workflows, you can efficiently build and manage highly customized, automated workflows.

Argo Workflows has a wide range of use cases, including scheduled tasks, machine learning, simulation computing, scientific computing, ETL, model training, and CI/CD. It primarily uses YAML to define workflows, a design choice intended to make configurations clear and concise. However, for users new to YAML or those designing complex workflows, its strict indentation requirements and hierarchical structure increase the learning curve and configuration complexity.

The Hera Python SDK offers the following advantages:

• Simplicity: Hera lets you write clear, understandable code, which improves development efficiency.

• Support for complex workflows: Hera helps you avoid the common syntax errors of YAML, especially in complex workflows.

• Python ecosystem integration: Each function becomes a template, allowing for easy integration with frameworks in the Python ecosystem.

• Testability: You can directly use Python's testing frameworks, which helps improve code quality and maintainability.

The following figure shows the architecture:

Step 1: Create a cluster and get a token

1. Create a distributed Argo workflow cluster.

2. Enable Argo Server to access the workflow cluster. You can do this in one of two ways:

   • Enable Argo Server to access the workflow cluster.

   • Enable public access to Argo Server. (Optional for users with leased lines)

3. Run the following command to create and get a cluster token.

   kubectl create token default -n default

Step 2: Use the Hera Python SDK

1. Run the following command to install Hera.

   pip install hera-workflows

2. Write and submit workflows.

   Scenario 1: Simple DAG diamond

   In Argo Workflows, a directed acyclic graph (DAG) is often used to define complex task dependencies. The diamond structure is a common workflow pattern that allows a task to be followed by multiple parallel tasks, which then converge into a single final task. This structure is effective for merging data streams or processing results. The following example uses Hera to define a workflow with a diamond structure, where tasks B and C run in parallel after task A completes, and their outputs are passed as a common input to task D.

   1. Create a file named simpleDAG.py with the following content.

      # Import the required packages.
      from hera.workflows import DAG, Workflow, script
      from hera.shared import global_config
      import urllib3
      urllib3.disable_warnings()
      # Configure the host address and token.
      global_config.host = "https://argo.{{clusterid}}.{{region-id}}.alicontainer.com:2746"
      global_config.token = "abcdefgxxxxxx"  # Enter the token that you obtained.
      global_config.verify_ssl = ""
      # The @script decorator is a key feature in Hera that enables near-native Python function orchestration.
      # It allows you to call the function within a Hera context manager, such as a Workflow or Steps context.
      # The function also runs normally outside of any Hera context, which means you can write unit tests for it.
      # This example prints the input message.
      @script()
      def echo(message: str):
          print(message)
      # Build the workflow. Workflow is a primary resource in Argo and a key class in Hera, 
      # responsible for managing templates, setting the entrypoint, and executing the workflow.
      with Workflow(
          generate_name="dag-diamond-",
          entrypoint="diamond",
          namespace="default",
      ) as w:
          with DAG(name="diamond"):
              A = echo(name="A", arguments={"message": "A"})  # Build a template.
              B = echo(name="B", arguments={"message": "B"})
              C = echo(name="C", arguments={"message": "C"})
              D = echo(name="D", arguments={"message": "D"})
              A >> [B, C] >> D      # Define the dependencies. Tasks B and C depend on A, and task D depends on B and C.
      # Create the workflow.
      w.create()

   2. Run the following command to submit the workflow.

      python simpleDAG.py

   3. After the workflow runs, you can view the task DAG and results in the Workflow Console (Argo).

      On the WORKFLOW DETAILS page, the workflow dag-diamond-g9v45 displays a diamond-shaped execution topology: task A is followed by parallel tasks B and C, which then converge into task D. A green checkmark on each node indicates that it executed successfully.

   Scenario 2: Map-Reduce

   Implementing Map-Reduce style data processing in Argo Workflows involves using its directed acyclic graph (DAG) template to organize and coordinate multiple tasks, simulating the Map and Reduce stages. The following example shows how to use Hera to build a simple Map-Reduce workflow. Each step is a Python function, which allows for easy integration with the Python ecosystem.

   1. Create a file named map-reduce.py with the following content.

      Expand to view the code

      from hera.workflows import DAG, Artifact, NoneArchiveStrategy, Parameter, OSSArtifact, Workflow, script
      from hera.shared import global_config
      import urllib3
      urllib3.disable_warnings()
      # Set the host address.
      global_config.host = "https://argo.{{clusterid}}.{{region-id}}.alicontainer.com:2746"
      global_config.token = "abcdefgxxxxxx"  # Enter the token that you obtained.
      global_config.verify_ssl = ""
      # When using the @script decorator, pass script parameters like image, inputs, outputs, and resources to the decorator.
      @script(
          image="python:alpine3.6",
          inputs=Parameter(name="num_parts"),
          outputs=OSSArtifact(name="parts", path="/mnt/out", archive=NoneArchiveStrategy(), key="{{workflow.name}}/parts"),
      )
      def split(num_parts: int) -> None:  # Creates multiple files based on the num_parts input parameter. Writes a JSON object with a 'foo' key and a part number to each file.
          import json
          import os
          import sys
          os.mkdir("/mnt/out")
          part_ids = list(map(lambda x: str(x), range(num_parts)))
          for i, part_id in enumerate(part_ids, start=1):
              with open("/mnt/out/" + part_id + ".json", "w") as f:
                  json.dump({"foo": i}, f)
          json.dump(part_ids, sys.stdout)
      # Define the image, inputs, and outputs in the script.
      @script(
          image="python:alpine3.6",
          inputs=[Parameter(name="part_id", value="0"), Artifact(name="part", path="/mnt/in/part.json"),],
          outputs=OSSArtifact(
              name="part",
              path="/mnt/out/part.json",
              archive=NoneArchiveStrategy(),
              key="{{workflow.name}}/results/{{inputs.parameters.part_id}}.json",
          ),
      )
      def map_() -> None:  # Creates a new file based on the value of the 'foo' key in the input file. It multiplies the 'foo' value by 2 and writes the result to a 'bar' key.
          import json
          import os
          os.mkdir("/mnt/out")
          with open("/mnt/in/part.json") as f:
              part = json.load(f)
          with open("/mnt/out/part.json", "w") as f:
              json.dump({"bar": part["foo"] * 2}, f)
      # Define the image, inputs, outputs, and resources in the script.
      @script(
          image="python:alpine3.6",
          inputs=OSSArtifact(name="results", path="/mnt/in", key="{{workflow.name}}/results"),
          outputs=OSSArtifact(
              name="total", path="/mnt/out/total.json", archive=NoneArchiveStrategy(), key="{{workflow.name}}/total.json"
          ),
      )
      def reduce() -> None:   # Calculate the sum of all 'bar' values from each part.
          import json
          import os
          os.mkdir("/mnt/out")
          total = 0
          for f in list(map(lambda x: open("/mnt/in/" + x), os.listdir("/mnt/in"))):
              result = json.load(f)
              total = total + result["bar"]
          with open("/mnt/out/total.json", "w") as f:
              json.dump({"total": total}, f)
      # Build the workflow. Set the name, entrypoint, namespace, and global parameters.
      with Workflow(generate_name="map-reduce-", entrypoint="main", namespace="default", arguments=Parameter(name="num_parts", value="4")) as w:
          with DAG(name="main"):
              s = split(arguments=Parameter(name="num_parts", value="{{workflow.parameters.num_parts}}")) # Build templates.
              m = map_(
                  with_param=s.result,
                  arguments=[Parameter(name="part_id", value="{{item}}"), OSSArtifact(name="part", key="{{workflow.name}}/parts/{{item}}.json"),],
              )   # Provide input parameters and build templates.
              s >> m >> reduce()   # Define the task dependencies.
      # Create the workflow.
      w.create()
      

   2. Run the following command to submit the workflow.

      python map-reduce.py

   3. After the workflow runs, the WORKFLOW DETAILS page in the Workflow Console (Argo) shows that all nodes in the map-reduce workflow (the split node, four parallel map nodes, and the reduce node) have completed successfully.

Comparison of authoring methods

The primary methods for authoring Argo Workflows are using YAML and the Hera framework. The following table compares these two approaches.

The Hera framework elegantly combines the Python ecosystem with Argo Workflows, making the design of complex workflows intuitive and straightforward. It provides a path for large-scale task orchestration that avoids the complexities of YAML, letting data engineers use the familiar Python language. This makes building and optimizing machine learning workflows seamless and efficient, accelerating the iteration cycle from idea to deployment and driving the rapid development and continuous evolution of intelligent applications.

If you have any questions about ACK One, join our DingTalk group by searching for the group ID: 35688562.

References

• Hera documentation

  • For detailed information about Hera, see the Hera overview.

  • To learn how to set up and use Hera for LLM training, see Train LLM with Hera.

• YAML deployment examples

  • To learn how to deploy a simple diamond workflow using YAML, see dag-diamond.yaml.

  • To learn how to deploy a map-reduce workflow using YAML, see map-reduce.yaml.