Container Service for Kubernetes:BatchCompute ジョブを分散型 Argo ワークフロークラスターへ移行

バッチコンピューティング

バッチコンピューティングには、線形の依存関係チェーンを形成する 5 つのコアコンポーネントがあります：計算環境はインフラストラクチャをホストし、ジョブキューは計算環境間でワークロードをルーティングし、ジョブ定義はジョブの実行方法を指定し、ジョブ（または配列ジョブ）は実際に提出・実行される作業単位です。

Argo Workflows

Argo Workflows には 3 つのコア構成要素があります。テンプレートは最小の作業単位を定義します。ワークフローは 1 つ以上のテンプレートをオーケストレーションします。ワークフローテンプレートは、複数のワークフロー間で再利用可能な静的なワークフロー定義を提供します。すべてのタスクは最終的に ACK サーバーレスクラスター上の Pod で実行されます。

シンプルなワークフロー

以下のワークフローは、alpine イメージを使用する Pod を作成し、シェルコマンド echo helloworld を実行します。

このワークフローをカスタマイズして、任意のシェルコマンドを実行したり、カスタムイメージ内のコマンドを実行したりできます。

cat > helloworld.yaml << EOF
apiVersion: argoproj.io/v1alpha1
kind: Workflow                  # 新しい k8s 仕様の種類
metadata:
  generateName: hello-world-    # ワークフロー仕様の名前
spec:
  entrypoint: main         # main テンプレートを呼び出します
  templates:
    - name: main              # テンプレートの名前
      container:
        image: registry.cn-hangzhou.aliyuncs.com/acs/alpine:3.18-update
        command: [ "sh", "-c" ]
        args: [ "echo helloworld" ]
EOF
argo submit helloworld.yaml

ループ

大量のデータを並列処理する場合にループを使用します — 各反復は独立した Pod として実行されます。

以下の例では、pets.input という名前のテキストファイルと print-pet.sh という名前のスクリプトをカスタムイメージにパッケージ化しています。ループは 5 つの Pod を作成し、job-index の値として 1 ～ 5 を入力パラメーターとして渡します。各 Pod は pets.input の対応する行に記載されたペットを出力します。ソースファイルの詳細については、GitHub リポジトリをご参照ください。

apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: loops-
spec:
  entrypoint: loop-example
  templates:
  - name: loop-example
    steps:
    - - name: print-pet
        template: print-pet
        arguments:
          parameters:
          - name: job-index
            value: "{{item}}"
        withSequence:  # print-pet テンプレートを、job-index パラメーターの値を 1 ～ 5 としてそれぞれ実行するループ。
          start: "1"
          end: "5"
  - name: print-pet
    inputs:
      parameters:
      - name: job-index
    container:
      image: acr-multiple-clusters-registry.cn-hangzhou.cr.aliyuncs.com/ack-multiple-clusters/print-pet
      command: [/tmp/print-pet.sh]
      args: ["{{inputs.parameters.job-index}}"] # コンテナの引数として入力パラメーター job-index を渡します

詳細については、「Argo Workflows - ループ」をご参照ください。

DAG（MapReduce）

バッチコンピューティングシステムでは、ジョブの依存関係をジョブ ID を用いて指定する必要があります — しかしジョブ ID はジョブが提出された後にのみ利用可能になります。典型的な回避策は、以下のようなスクリプトです：

// ジョブ B はジョブ A に依存。ジョブ A の完了後にジョブ B が開始されます。
batch submit JobA | get job-id
batch submit JobB --dependency job-id (JobA)

ジョブ数が増えるにつれて、このような依存関係スクリプトのメンテナンスは困難になります。

Argo Workflows では、ワークフロー定義内に直接タスク名で依存関係を宣言できます：

• タスク B およびタスク C はタスク A に依存します。

• タスク D はタスク B およびタスク C に依存します。

# 以下のワークフローはダイヤモンド型ワークフローを実行します
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  generateName: dag-diamond-
spec:
  entrypoint: diamond
  templates:
  - name: diamond
    dag:
      tasks:
      - name: A
        template: echo
        arguments:
          parameters: [{name: message, value: A}]
      - name: B
        depends: "A"
        template: echo
        arguments:
          parameters: [{name: message, value: B}]
      - name: C
        depends: "A"
        template: echo
        arguments:
          parameters: [{name: message, value: C}]
      - name: D
        depends: "B && C"
        template: echo
        arguments:
          parameters: [{name: message, value: D}]
  - name: echo
    inputs:
      parameters:
      - name: message
    container:
      image: alpine:3.7
      command: [echo, "{{inputs.parameters.message}}"]

シャードを作成して結果を集約するサンプル MapReduce ワークフローについては、「map-reduce」をご参照ください。

ステップ 1：評価と計画

ジョブの変換を開始する前に、既存のバッチワークロードを以下の観点で包括的に評価してください：

• ジョブの在庫調査：すべてのジョブ（依存関係、リソース要求（CPU、メモリ、GPU）、実行時パラメーターを含む）をリストアップします。

• 優先順位付け：ビジネス上重大なジョブと高複雑度のジョブを別々に特定します。まずシンプルで非重要なジョブから移行を開始し、複雑なワークフローに着手する前に信頼性を確立してください。

• 機能ギャップの確認：上記の機能対応表を用いて、各バッチコンピューティング機能を Argo Workflows の対応機能にマッピングします。計算環境設計およびジョブ優先度設定はスキップしてください — 分散型 Argo ワークフロー向けの Kubernetes クラスターはサーバーレスのエラスティックコンテナインスタンスを使用し、キュー管理を必要としません。

ステップ 2：分散型 Argo ワークフロー向けの Kubernetes クラスターの作成

「ワークフロークラスターの作成」をご参照ください。

ステップ 3：ジョブ定義の変換

機能対応表に基づき、バッチジョブを Argo ワークフローに変換します。必要に応じて、Argo Workflows SDK を使用して、ワークフロー作成およびシステム統合を自動化できます。

ステップ 4：ストレージの準備

クラスターがワークフローで必要なデータにアクセスできるよう、事前に準備を行います。OSS バケット、NAS ファイルシステム、CPFS ファイルシステム、またはディスクをクラスターにマウントします。手順については、「ボリュームの使用」をご参照ください。

ステップ 5：ワークフローの検証

バッチシステムを廃止する前に、両方のシステムを並行して実行します：

1. 既存のバッチジョブは無効化したまま、比較用に利用可能にしておきます。

2. 同一の入力データに対して、両方のシステムで同一のワークロードを実行します。

3. 出力を競合しないよう、別々のパスに書き込みます。

4. 結果（データ精度、出力構造、リソース使用量）を比較し、同等性を確認します。

5. 正しさが確認された後、バッチジョブを廃止します。

並行実行期間は短く保ってください。長期にわたる二重運用は、運用オーバーヘッドおよび乖離リスクを増加させます。

ステップ 6：監視およびログ記録の有効化

ワークフローのステータスおよびログを確認するため、クラスターの可観測性を有効化します。さらに計算コストを削減するため、プリエンプティブルインスタンスを利用できます。