Container Service for Kubernetes:Membangun Argo Workflows berskala besar dengan Python SDK

Latar Belakang

Argo Workflows adalah tool open-source native Kubernetes untuk mengorkestrasi workflow yang kompleks dengan dependensi tugas yang fleksibel.

Kasus penggunaannya mencakup tugas terjadwal, machine learning, simulasi, komputasi ilmiah, ETL, pelatihan model, dan CI/CD. Argo Workflows menggunakan YAML untuk mendefinisikan workflow, tetapi indentasi yang ketat dan struktur hierarkis YAML menciptakan kurva belajar yang curam untuk workflow yang kompleks.

Hera adalah Python SDK untuk membangun dan mengirimkan Argo Workflows, yang memungkinkan data scientist bekerja dalam Python, bukan YAML.

Membuat kluster dan mendapatkan token

1. Buat kluster alur kerja Argo, lalu aktifkan Argo Server dan akses konsol alur kerja.

2. Buat token kluster.

   kubectl create token default -n default

Mengirimkan workflow dengan Hera

1. Instal Hera.

   pip install hera

2. Tulis dan kirimkan workflow.

   DAG diamond sederhana

   DAG diamond adalah pola umum di mana tugas bercabang dan kemudian konvergen. Dalam contoh ini, Task A berjalan terlebih dahulu, kemudian B dan C berjalan secara paralel, dan Task D berjalan setelah keduanya selesai.

   1. Buat file bernama simpleDAG.py dengan konten berikut.

      # Impor paket yang diperlukan.
      from hera.workflows import DAG, Workflow, script
      from hera.shared import global_config
      import urllib3
      urllib3.disable_warnings()
      # Konfigurasikan alamat host dan token.
      global_config.host = "https://{{argo_server_IP}}:2746"
      global_config.token = "abcdefgxxxxxx"  # Ganti dengan token yang Anda peroleh.
      global_config.verify_ssl = ""
      # Dekorator @script adalah fitur utama Hera yang memungkinkan Anda mengorkestrasi fungsi Python hampir secara native.
      # Ini memungkinkan Anda memanggil fungsi yang didekorasi dalam context manager Hera, seperti konteks Workflow atau Steps.
      # Fungsi ini tetap berjalan normal di luar konteks Hera apa pun, yang berarti Anda dapat menulis unit test untuknya.
      # Contoh ini mencetak pesan input.
      @script(image="mirrors-ssl.aliyuncs.com/python:3.10")
      def echo(message: str):
          print(message)
      # Workflow adalah resource utama dalam Argo dan kelas kunci dalam Hera. Ini menyimpan templat, menetapkan entrypoint, dan menjalankannya.
      with Workflow(
          generate_name="dag-diamond-",
          entrypoint="diamond",
          namespace="default",
      ) as w:
          with DAG(name="diamond"):
              A = echo(name="A", arguments={"message": "A"})  # Bangun templat.
              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      # Tentukan dependensi: Task B dan C bergantung pada A, dan task D bergantung pada B dan C.
      # Buat workflow.
      w.create()

   2. Kirimkan workflow.

      python simpleDAG.py

   3. Lihat tugas DAG dan hasilnya di workflow console.

      Workflow dag-diamond-g9v45 menampilkan DAG diamond: node A selesai, B dan C berjalan secara paralel, lalu konvergen di D. Semua node menunjukkan eksekusi yang berhasil.

   Map-reduce

   Contoh ini menggunakan templat DAG untuk mengimplementasikan MapReduce: memecah tugas menjadi langkah map paralel, lalu menggabungkan hasilnya dalam langkah reduce. Setiap langkah merupakan fungsi Python.

   1. Konfigurasikan artefak.

   2. Buat file bernama map-reduce.py dengan konten berikut.

      Kode

      from hera.workflows import DAG, Artifact, NoneArchiveStrategy, Parameter, OSSArtifact, Workflow, script
      from hera.shared import global_config
      import urllib3
      urllib3.disable_warnings()
      # Atur alamat host.
      global_config.host = "https://{{argo_server_IP}}:2746"
      global_config.token = "abcdefgxxxxxx"  # Ganti dengan token yang Anda peroleh.
      global_config.verify_ssl = ""
      # Saat menggunakan dekorator @script, berikan parameter seperti image, inputs, outputs, dan resources kepadanya.
      @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:  # Fungsi ini membuat beberapa file output berdasarkan parameter input num_parts. Ini menulis bidang 'foo' dan nomor bagian ke setiap 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)
      # Tentukan image, inputs, dan outputs dalam dekorator @script.
      @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:  # Fungsi ini membaca nilai 'foo' dari file input, mengalikannya dengan 2, dan menulis hasilnya ke bidang 'bar' dalam file output baru.
          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)
      # Tentukan image, inputs, outputs, dan resources dalam dekorator @script.
      @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:   # Fungsi ini menghitung jumlah nilai 'bar' dari semua task map.
          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)
      # Bangun workflow. Tentukan nama, entrypoint, namespace, dan parameter globalnya.
      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}}")) # Bangun templat.
              m = map_(
                  with_param=s.result,
                  arguments=[Parameter(name="part_id", value="{{item}}"), OSSArtifact(name="part", key="{{workflow.name}}/parts/{{item}}.json"),],
              )   # Berikan parameter dan bangun templat.
              s >> m >> reduce()   # Tentukan dependensi task.
      # Buat workflow.
      w.create()
      

   3. Kirimkan workflow.

      python map-reduce.py

   4. Lihat tugas DAG dan hasilnya di workflow console. Pada halaman WORKFLOW DETAILS, DAG menunjukkan node split, empat node map paralel, dan node reduce yang semuanya berhasil dieksekusi (tanda centang hijau).

Referensi

• Dokumentasi Hera:

  • Ikhtisar Hera

  • Melatih LLM dengan Hera

• Contoh deployment YAML:

  • Diamond sederhana (YAML): dag-diamond.yaml

  • MapReduce (YAML): map-reduce.yaml