Python SDK で大規模な Argo Workflows を構築 - Container Service for Kubernetes

Argo Workflows は、定期タスク、機械学習、ETL に広く使用されていますが、YAML の定義は習得が難しい場合があります。Hera Python SDK を使用すると、Python でワークフローを構築でき、複雑なタスクのサポート、テストの簡素化、Python エコシステムとのシームレスな統合が可能になります。このトピックでは、Python SDK を使用して大規模な Argo Workflows を構築する方法について説明します。

背景情報

Argo Workflows は、オープンソースの Kubernetes ネイティブツールであり、柔軟なタスク依存関係を持つ複雑なワークフローをオーケストレーションします。

ユースケースには、定期タスク、機械学習、シミュレーション、科学計算、ETL、モデルトレーニング、CI/CD などが含まれます。Argo Workflows は YAML を使用してワークフローを定義しますが、YAML の厳密なインデントと階層構造は、複雑なワークフローを扱う際の習得を難しくしています。

Hera は、Argo Workflows を構築およびサブミットするための Python SDK であり、データサイエンティストは YAML の代わりに Python で作業できます。

作成方法の比較

	YAML	Hera
シンプルさ	高	高、コード行数が少ない
複雑なワークフローの作成	困難	簡単。YAML の構文エラーを回避できます。
Python エコシステムとの統合	困難	簡単。Python ライブラリにアクセスできます。
テスト容易性	困難、構文エラーが発生しやすい	簡単。テストフレームワークにより、コードの品質と保守性が向上します。

Hera は Python エコシステムと Argo Workflows の橋渡しをし、YAML の複雑さを伴わずに大規模なタスクのオーケストレーションを可能にします。以下の例では Hera を使用します。

クラスターの作成とトークンの取得

Argo ワークフロークラスターを作成し、次に Argo Server を有効にしてワークフローコンソールにアクセスします。
クラスタートークンを作成します。
```
kubectl create token default -n default
```

Hera を使用したワークフローのサブミット

Hera をインストールします。
```
pip install hera
```

ワークフローを作成してサブミットします。

シンプルな DAG ダイヤモンド

DAG ダイヤモンドは、タスクが分岐し、再び合流する一般的なパターンです。この例では、タスク A が最初に実行され、次に B と C が並行して実行され、両方が完了した後にタスク D が実行されます。

simpleDAG.py という名前のファイルを作成し、次の内容を記述します。

# 必要なパッケージをインポートします。
from hera.workflows import DAG, Workflow, script
from hera.shared import global_config
import urllib3
urllib3.disable_warnings()
# ホストアドレスとトークンを設定します。
global_config.host = "https://{{argo_server_IP}}:2746"
global_config.token = "abcdefgxxxxxx"  # 取得したトークンに置き換えます。
global_config.verify_ssl = False
# @script デコレーターは Hera の重要な機能で、Python 関数をネイティブに近い形でオーケストレーションできます。
# これにより、Workflow や Steps のコンテキストマネージャーなど、Hera のコンテキスト内でデコレートされた関数を呼び出すことができます。
# この関数は、Hera のコンテキスト外でも通常通り実行されるため、ユニットテストを作成できます。
# この例では、入力メッセージを出力します。
@script(image="mirrors-ssl.aliyuncs.com/python:3.10")
def echo(message: str):
    print(message)
# Workflow は Argo の主要なリソースであり、Hera の主要なクラスです。テンプレートを保存し、エントリポイントを設定して実行します。
with Workflow(
    generate_name="dag-diamond-",
    entrypoint="diamond",
    namespace="default",
) as w:
    with DAG(name="diamond"):
        A = echo(name="A", arguments={"message": "A"})  # テンプレートを構築します。
        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      # 依存関係を定義します：タスク B と C は A に依存し、タスク D は B と C に依存します。
# ワークフローを作成します。
w.create()

ワークフローをサブミットします。
```
python simpleDAG.py
```
[ワークフローコンソール] でタスクの DAG と結果を表示します。

ワークフロー dag-diamond-g9v45 は、ダイヤモンド型の DAG を示しています。ノード A が完了し、B と C が並行して実行され、D で合流します。すべてのノードが正常に実行されたことを示しています。

MapReduce

この例では、DAG テンプレートを使用して MapReduce を実装します。タスクを並列の map ステップに分割し、その結果を reduce ステップで集約します。各ステップは Python の関数です。

アーティファクトを設定します。

map-reduce.py という名前のファイルを作成し、次の内容を記述します。

コード

from hera.workflows import DAG, Artifact, NoneArchiveStrategy, Parameter, OSSArtifact, Workflow, script
from hera.shared import global_config
import urllib3
urllib3.disable_warnings()
# ホストアドレスを設定します。
global_config.host = "https://{{argo_server_IP}}:2746"
global_config.token = "abcdefgxxxxxx"  # 取得したトークンに置き換えます。
global_config.verify_ssl = False
# @script デコレーターを使用する場合、image、inputs、outputs、resources などのパラメーターを渡します。
@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:  # この関数は、num_parts 入力パラメーターに基づいて複数の出力ファイルを作成します。「foo」フィールドと 1 から始まる連番を各ファイルに書き込みます。
    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)
# @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:  # この関数は、入力ファイルから「foo」の値を読み取り、それを 2 倍して、新しい出力ファイルの「bar」フィールドに書き込みます。
    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)
# @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:   # この関数は、すべての map タスクからの「bar」の値の合計を計算します。
    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)
# ワークフローを構築します。名前、エントリポイント、名前空間、グローバルパラメーターを定義します。
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}}")) # テンプレートを構築します。
        m = map_(
            with_param=s.result,
            arguments=[Parameter(name="part_id", value="{{item}}"), OSSArtifact(name="part", key="{{workflow.name}}/parts/{{item}}.json"),],
        )   # パラメーターを渡し、テンプレートを構築します。
        s >> m >> reduce()   # タスクの依存関係を定義します。
# ワークフローを作成します。
w.create()

ワークフローをサブミットします。
```
python map-reduce.py
```
[ワークフローコンソール] でタスクの DAG と結果を表示します。[WORKFLOW DETAILS] ページで、DAG には split ノード、4 つの並列 map ノード、および reduce ノードがすべて正常に実行されたこと (緑のチェックマーク) が示されています。

Container Service for Kubernetes:Python SDK で大規模な Argo Workflows を構築