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

Contexto

O Argo Workflows é uma ferramenta open source nativa do Kubernetes para orquestrar workflows complexos com dependências flexíveis entre tarefas.

Os casos de uso incluem tarefas agendadas, machine learning, simulação, computação científica, ETL, treinamento de modelos e CI/CD. O Argo Workflows utiliza YAML para definir workflows, mas a indentação rigorosa e a estrutura hierárquica do YAML geram uma curva de aprendizado íngreme para workflows complexos.

O Hera é um SDK para Python destinado à criação e ao envio de Argo Workflows, permitindo que cientistas de dados trabalhem em Python em vez de YAML.

Crie um cluster e obtenha um token

1. Create an Argo workflow cluster e, em seguida, ative Argo Server and access the workflow console.

2. Crie um token de cluster.

   kubectl create token default -n default

Envie workflows com o Hera

1. Instale o Hera.

   pip install hera

2. Escreva e envie os workflows.

   Diamante DAG simples

   Um diamante DAG é um padrão comum em que as tarefas divergem e convergem. Neste exemplo, a Tarefa A executa primeiro; depois, B e C executam em paralelo; por fim, a Tarefa D executa após a conclusão de ambas.

   1. Crie um file chamado simpleDAG.py com o conteúdo a seguir.

      # 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_server_IP}}:2746"
      global_config.token = "abcdefgxxxxxx"  # Replace with the token you obtained.
      global_config.verify_ssl = ""
      # The @script decorator is a key Hera feature that lets you orchestrate Python functions almost natively.
      # It allows you to call the decorated function within a Hera context manager, such as a Workflow or Steps context.
      # The function still runs normally outside of any Hera context, which means you can write unit tests for it.
      # This example prints the input message.
      @script(image="mirrors-ssl.aliyuncs.com/python:3.10")
      def echo(message: str):
          print(message)
      # A Workflow is the primary resource in Argo and a key class in Hera. It stores templates, sets an entrypoint, and runs them.
      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. Envie o workflow.

      python simpleDAG.py

   3. Visualize o DAG da tarefa e os resultados no workflow console.

      O workflow dag-diamond-g9v45 exibe um DAG em formato de diamante: o nó A é concluído, B e C executam em paralelo e depois convergem em D. Todos os nós mostram execução bem-sucedida.

   Map-reduce

   Este exemplo usa modelos DAG para implementar MapReduce: divide uma tarefa em etapas de map paralelas e agrega os resultados em uma etapa de reduce. Cada etapa é uma função Python.

   1. Configure artifacts.

   2. Crie um file chamado map-reduce.py com o conteúdo a seguir.

      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_server_IP}}:2746"
      global_config.token = "abcdefgxxxxxx"  # Replace with the token you obtained.
      global_config.verify_ssl = ""
      # When using the @script decorator, pass parameters like image, inputs, outputs, and resources to it.
      @script(
          image="mirrors-ssl.aliyuncs.com/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:  # This function creates multiple output files based on the num_parts input parameter. It writes a 'foo' field 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 decorator.
      @script(
          image="mirrors-ssl.aliyuncs.com/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:  # This function reads the 'foo' value from an input file, multiplies it by 2, and writes the result to a 'bar' field in a new output file.
          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 decorator.
      @script(
          image="mirrors-ssl.aliyuncs.com/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:   # This function calculates the sum of the 'bar' values from all the map tasks.
          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. Define its 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 the templates.
              m = map_(
                  with_param=s.result,
                  arguments=[Parameter(name="part_id", value="{{item}}"), OSSArtifact(name="part", key="{{workflow.name}}/parts/{{item}}.json"),],
              )   # Pass parameters and build the templates.
              s >> m >> reduce()   # Define the task dependencies.
      # Create the workflow.
      w.create()
      

   3. Envie o workflow.

      python map-reduce.py

   4. Visualize o DAG da tarefa e os resultados no workflow console. Na página WORKFLOW DETAILS, o DAG mostra o nó split, quatro nós map paralelos e o nó reduce, todos executados com sucesso (marca de verificação verde).

Referências

• Documentação do Hera:

  • Visão geral do Hera

  • Treine LLM com Hera

• Exemplos de implantação em YAML:

  • Diamante simples (YAML): dag-diamond.yaml

  • Map-reduce (YAML): map-reduce.yaml