Container Service for Kubernetes:Construcción de flujos de trabajo Argo a gran escala con el SDK de Python

Contexto

Argo Workflows es una herramienta de gestión de flujos de trabajo de código abierto diseñada específicamente para entornos de Kubernetes. Se centra en la orquestación de flujos de trabajo complejos y permite definir una serie de tareas, además de organizar con flexibilidad su orden de ejecución y sus dependencias. Argo Workflows le ayuda a construir y gestionar de manera eficiente flujos de trabajo automatizados y altamente personalizados.

Argo Workflows tiene una amplia variedad de casos de uso, entre ellos tareas programadas, aprendizaje automático, cómputo por simulación, cómputo científico, ETL, entrenamiento de modelos y CI/CD. Utiliza principalmente YAML para definir los flujos de trabajo, una decisión de diseño orientada a la claridad y la simplicidad. No obstante, para quienes no tienen experiencia en YAML, su estricta sangría y estructura jerárquica pueden generar una curva de aprendizaje pronunciada, especialmente en flujos de trabajo complejos.

Hera es un framework del SDK de Python diseñado para construir y enviar flujos de trabajo de Argo. Simplifica tanto la creación como el envío de estos flujos. Para los científicos de datos, usar Python se alinea con sus prácticas habituales y les ayuda a superar los desafíos que presenta YAML.

Paso 1: Crear un clúster y obtener un token

1. Create an Argo workflow cluster, luego enable Argo Server and access the workflow console.

2. Cree un token de clúster.

   kubectl create token default -n default

Paso 2: Enviar flujos de trabajo con Hera

1. Instale Hera.

   pip install hera

2. Escriba y envíe los flujos de trabajo.

   Diamante DAG simple

   En Argo Workflows, un DAG (grafo acíclico dirigido) se usa frecuentemente para definir dependencias complejas entre tareas. El patrón de diamante es común: las tareas divergen y luego convergen. Esta estructura resulta eficaz para el procesamiento paralelo donde los resultados se agregan en una tarea posterior común. El siguiente ejemplo muestra cómo usar Hera para definir un flujo de trabajo con estructura de diamante. La tarea A se ejecuta primero, seguida de dos tareas paralelas, B y C. Una última tarea, D, se ejecuta después de que terminan tanto B como C, y completa así el flujo de trabajo.

   1. Cree un archivo llamado simpleDAG.py con el siguiente contenido.

      # 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. Envíe el flujo de trabajo.

      python simpleDAG.py

   3. Una vez que se ejecute el flujo de trabajo, podrá consultar el DAG de tareas y sus resultados en la workflow console.

      El flujo de trabajo de ejemplo dag-diamond-g9v45 muestra una topología DAG en forma de diamante: el nodo de nivel superior A se completa, luego los nodos B y C se ejecutan en paralelo y, finalmente, convergen en el nodo D. Todos los nodos aparecen marcados como ejecutados correctamente.

   Map-reduce

   En Argo Workflows, es posible implementar procesamiento de datos estilo MapReduce mediante plantillas DAG para simular las fases map y reduce. El siguiente ejemplo demuestra cómo usar Hera para construir un flujo de trabajo sencillo de MapReduce. Este flujo divide una tarea en varias tareas map paralelas y luego agrega sus resultados en una tarea final de reduce. Cada paso es una función de Python, lo que facilita la integración con el ecosistema de Python.

   1. Configure artifacts.

   2. Cree un archivo llamado map-reduce.py con el siguiente contenido.

      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. Envíe el flujo de trabajo.

      python map-reduce.py

   4. Después de que se ejecute el flujo de trabajo, podrá consultar el DAG de tareas y sus resultados en la workflow console. En la página WORKFLOW DETAILS, la vista DAG muestra que el nodo split, los cuatro nodos map paralelos y el nodo reduce se han ejecutado correctamente (indicado con una marca de verificación verde).

Referencias

• Documentación de Hera:

  • Obtenga más información sobre Hera en la descripción general de Hera.

  • Para entrenar un modelo de lenguaje grande (LLM) con Hera, consulte Entrenamiento de LLM con Hera.

• Ejemplos de despliegue con YAML:

  • Para desplegar el ejemplo simple de diamante mediante YAML, consulte dag-diamond.yaml.

  • Para desplegar el ejemplo de map-reduce mediante YAML, consulte map-reduce.yaml.