Container Service for Kubernetes:Bangun Argo Workflows skala besar dengan Python SDK

Latar Belakang

Argo Workflows adalah alat manajemen alur kerja open-source yang dirancang khusus untuk lingkungan Kubernetes. Fokus utamanya adalah mengorkestrasikan alur kerja kompleks dengan memungkinkan pengguna mendefinisikan serangkaian Tugas serta mengatur urutan eksekusi dan dependensinya secara fleksibel. Argo Workflows membantu Anda membangun dan mengelola alur kerja otomatis yang sangat disesuaikan secara efisien.

Argo Workflows memiliki berbagai kasus penggunaan, termasuk tugas terjadwal, pembelajaran mesin, komputasi simulasi, komputasi ilmiah, ETL, pelatihan model, dan CI/CD. Alur kerja utamanya didefinisikan menggunakan YAML—pilihan desain yang ditujukan untuk kejelasan dan kesederhanaan. Namun, bagi pengguna baru atau yang tidak terbiasa dengan YAML, struktur hierarkis dan indentasi ketatnya dapat menciptakan kurva belajar yang curam, terutama untuk alur kerja yang kompleks.

Hera adalah framework Python SDK yang dirancang untuk membangun dan mengirimkan alur kerja Argo. Framework ini menyederhanakan pembuatan dan pengiriman alur kerja. Bagi ilmuwan data, penggunaan Python selaras dengan praktik umum mereka dan membantu mengatasi tantangan penggunaan YAML.

Langkah 1: Buat kluster dan dapatkan token

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

2. Buat token kluster.

   kubectl create token default -n default

Langkah 2: Kirim alur kerja dengan Hera

1. Instal Hera.

   pip install hera

2. Tulis dan kirim alur kerja.

   DAG diamond sederhana

   Dalam Argo Workflows, DAG (Directed Acyclic Graph) sering digunakan untuk mendefinisikan dependensi Tugas yang kompleks. Pola diamond adalah pola umum di mana Tugas bercabang lalu berkumpul kembali. Struktur ini efektif untuk pemrosesan paralel di mana hasilnya diagregasi ke Tugas downstream bersama. Contoh berikut menunjukkan cara menggunakan Hera untuk mendefinisikan alur kerja dengan struktur diamond: Tugas A dijalankan terlebih dahulu, diikuti oleh dua Tugas paralel, B dan C. Tugas akhir, D, dijalankan setelah B dan C selesai, menyelesaikan alur kerja.

   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 mengorkestrasikan fungsi Python hampir secara native.
      # Fitur ini memungkinkan Anda memanggil fungsi yang didekorasi dalam context manager Hera, seperti konteks Workflow atau Steps.
      # Fungsi tetap berjalan normal di luar konteks Hera apa pun, artinya Anda dapat menulis pengujian unit 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 di Argo dan kelas utama di Hera. Workflow 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      # Definisikan dependensi: Tugas B dan C bergantung pada A, dan tugas D bergantung pada B dan C.
      # Buat alur kerja.
      w.create()

   2. Kirim alur kerja.

      python simpleDAG.py

   3. Setelah alur kerja dijalankan, Anda dapat melihat DAG Tugas dan hasilnya di workflow console.

      Alur kerja contoh dag-diamond-g9v45 menunjukkan topologi DAG berbentuk diamond: node tingkat atas A selesai, lalu node B dan C dijalankan secara paralel, dan akhirnya berkumpul di node D. Semua node ditandai sebagai berhasil dieksekusi.

   Map-reduce

   Dalam Argo Workflows, Anda dapat menerapkan pemrosesan data bergaya MapReduce menggunakan templat DAG untuk mensimulasikan fase map dan reduce. Contoh berikut menunjukkan cara menggunakan Hera untuk membangun alur kerja MapReduce sederhana: alur kerja ini membagi Tugas menjadi beberapa Tugas map paralel, lalu mengagregasi hasilnya dalam satu Tugas reduce akhir. Setiap langkah merupakan fungsi Python, yang memungkinkan integrasi mudah dengan ekosistem 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()
      # Tetapkan 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 ke dalamnya.
      @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. Fungsi 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)
      # Definisikan 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, lalu 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)
      # Definisikan 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 tugas 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 alur kerja. Definisikan nama, entrypoint, namespace, dan parameter global-nya.
      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()   # Definisikan dependensi tugas.
      # Buat alur kerja.
      w.create()
      

   3. Kirim alur kerja.

      python map-reduce.py

   4. Setelah alur kerja dijalankan, Anda dapat melihat DAG Tugas dan hasilnya di workflow console. Di halaman WORKFLOW DETAILS, tampilan DAG menunjukkan bahwa node split, empat node map paralel, dan node reduce semuanya telah dieksekusi dengan sukses (ditandai dengan centang hijau).

Referensi

• Dokumentasi Hera:

  • Pelajari lebih lanjut tentang Hera di Hera Overview.

  • Untuk melatih model bahasa besar (LLM) dengan Hera, lihat Train LLM with Hera.

• Contoh penerapan YAML:

  • Untuk menerapkan contoh diamond sederhana menggunakan YAML, lihat dag-diamond.yaml.

  • Untuk menerapkan contoh map-reduce menggunakan YAML, lihat map-reduce.yaml.