PAI における DLC、DSW、EAS サービスのタギングの標準化 - Platform For AI

タグルールにより、ワークスペースの管理者は DLC、DSW、EAS 全体で一貫したタギングを徹底できます。管理者は、必須のタグキー、事前定義された値、説明、および適用対象の製品範囲を定義します。ルールが有効な場合、ML 開発者は DLC ジョブ、DSW インスタンス、EAS サービスの作成時にタギングを行う必要があり、タグは自動的に検証されます。一貫性のあるタグは、コスト配分、ジョブの分布分析、リソース消費レポートに役立ちます。

タグルールの定義

説明

ワークスペースの 所有者 と 管理者 のみがタグルールを編集できます。

ワークスペースの詳細ページに移動し、右上隅にある ワークスペースの詳細 > 一般設定 をクリックします。
[Tag Configuration] セクションにスクロールし、設定の編集 をクリックして編集モードに入ります。
必要に応じてタグルールを追加し、保存をクリックします。設定したルールは、選択した製品範囲で有効になります。対応するサブ製品ページでは、これらのタグがデフォルトで必須となります。タグが欠落しているか、ルールに準拠していない場合、送信は失敗します。ML 開発者は、タグキーとタグ値の要件が満たされている限り、必須のタグに加えてカスタムタグを追加することもできます。
- [タグキー]の要件: aliyun または acs: で始まらず、http:// または https:// を含まず、長さが 128 文字以下である必要があります。使用できるのは、英字、数字、ハイフン (-)、中国語文字、ピリオド (.) のみで、アンダースコア (_) は使用できません。
- 値の要件: http:// または https:// を含めることはできず、長さは 256 文字以下である必要があります。複数の列挙値は、コンマで区切ります。
- [値の制限]：このオプションを有効にすると、ユーザーは事前定義された値から選択する必要があり、カスタム値は入力できません。このオプションを有効にすると、チーム全体で一貫したタギングを徹底できます。
- [対象プロダクト]：このルールが適用されるサブ製品 (DLC、DSW、または EAS) を選択します。少なくとも 1 つを選択する必要があります。

タグの適用

DLC ジョブを送信するか、DSW または EAS インスタンスを作成する際に、ワークスペースのタグルールに従ってタグを適用します。以下の手順では、DLC ジョブの送信を例として説明します：

DLC コンソールに移動し、[Create Job] をクリックします。
ジョブ設定ページのタグセクションで、ワークスペースのタグルールに従ってキーと値のペアを入力します。

ジョブ作成時にカスタムタグを追加することもできます。
すべての必須タグを入力した後、ジョブを送信します。

重要

タグルールは送信時に検証されます。必須のタグが欠落しているか、値がルールと一致しない場合、送信は失敗し、エラーメッセージが表示されます。

タグによるジョブのフィルタリング

DLC ジョブリストでタグベースのフィルタリングを使用すると、シナリオ、モデルタイプ、またはその他の基準ごとにジョブの分布とリソース使用量を分析できます。

DLC コンソールのジョブリストページに移動します。
検索バーで、フィルター条件としてタグを選択し、タグキーと値を入力して [検索] をクリックします。リストには、基準に一致するジョブのみが表示されます。

付録

API リファレンス

タグルールを一括で設定したり、自動化パイプラインと統合したりするには、以下のコード例を使用します。コードを実行する前に、リージョン、ワークスペース ID、AccessKey ID と AccessKey Secret の値を置き換えてください。

タグルールの一括設定

UpdateConfigs API は、プリセットタグルールの一括設定をサポートし、複数のワークスペースに同じタグ設定を適用するのに適しています。

リクエストボディの構造例：

{
  "Configs": [
    {
      "ConfigKey": "predefinedTags",
      "ConfigValue": "[{\"Type\":\"Tag\",\"Key\":\"Scene\",\"Value\":\"{\\\"Products\\\":\\\"DLC,DSW,EAS\\\",\\\"Values\\\":\\\"Training,Evaluation,Inference\\\",\\\"ValueFlexible\\\":false,\\\"Description\\\":\\\"select scenario\\\"}\"}]",
      "Labels": [
        {
          "Key": "system.categoryName",
          "Value": "CommonTagConfig"
        }
      ]
    }
  ]
}

Python コード例：タグルールの一括設定

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PAI ワークスペース - 事前定義されたタグ ルール (CommonTagConfig) の一括作成/更新
API: PUT /api/v1/workspaces/{WorkspaceId}/configs
使用する前に、ACCESS_KEY_ID、ACCESS_KEY_SECRET、REGION_ID、および WORKSPACE_ID を入力し、
TAGS リストでタグ ルールを定義してください。
"""

import json, sys, hashlib, hmac, datetime, requests

ACCESS_KEY_ID     = "your-AccessKeyId"
ACCESS_KEY_SECRET = "your-AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "your-workspace-ID"

ENDPOINT = f"aiworkspace.{REGION_ID}.aliyuncs.com"
API_PATH = f"/api/v1/workspaces/{WORKSPACE_ID}/configs"

# タグの定義: key-タグ名、products-適用対象製品、values-許可される値、flexible-カスタム値が許可されるかどうか、desc-説明
TAGS = [
    {"key": "Scene",      "products": "DLC,DSW,EAS", "values": "Training,Evaluation,Inference",         "flexible": False, "desc": "現在の使用シナリオを選択"},
    {"key": "ModelType",  "products": "DLC,DSW",     "values": "NLP-Text,CV-Image,Multimodal",          "flexible": False, "desc": "モデルタイプを選択"},
    {"key": "ExpDirection","products": "DLC",         "values": "LR-Schedule,DataAugmentation,Distillation", "flexible": True, "desc": "実験の方向性をカスタマイズ"},
]

def build_config_value(tags):
    arr = []
    for t in tags:
        inner = {"Products": t["products"], "Values": t["values"],
                 "ValueFlexible": t.get("flexible", False), "Description": t.get("desc", "")}
        arr.append({"Type": "Tag", "Key": t["key"], "Value": json.dumps(inner, ensure_ascii=False)})
    return json.dumps(arr, ensure_ascii=False)

def sign_request(method, path, headers, body):
    now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    headers.update({"x-acs-date": now, "x-acs-version": "2021-02-04",
                    "x-acs-action": "UpdateConfigs",
                    "x-acs-signature-nonce": hashlib.md5((now + body).encode()).hexdigest()})
    signed_keys = sorted(k for k in headers if k.startswith("x-acs-") or k in ("host", "content-type"))
    canonical_headers = "".join(f"{k}:{headers[k]}\n" for k in signed_keys)
    hashed_body = hashlib.sha256(body.encode("utf-8")).hexdigest()
    canonical_request = f"{method}\n{path}\n\n{canonical_headers}\n{';'.join(signed_keys)}\n{hashed_body}"
    string_to_sign = f"ACS3-HMAC-SHA256\n{hashlib.sha256(canonical_request.encode()).hexdigest()}"
    signature = hmac.new(ACCESS_KEY_SECRET.encode(), string_to_sign.encode(), hashlib.sha256).hexdigest()
    headers["Authorization"] = (f"ACS3-HMAC-SHA256 Credential={ACCESS_KEY_ID},"
                                 f"SignedHeaders={';'.join(signed_keys)},Signature={signature}")
    return headers

def main():
    config_value = build_config_value(TAGS)
    request_body = json.dumps({"Configs": [{"ConfigKey": "predefinedTags", "ConfigValue": config_value,
                                            "Labels": [{"Key": "system.categoryName", "Value": "CommonTagConfig"}]}]},
                              ensure_ascii=False)
    url = f"https://{ENDPOINT}{API_PATH}"
    headers = sign_request("PUT", API_PATH, {"host": ENDPOINT, "content-type": "application/json"}, request_body)
    resp = requests.put(url, headers=headers, data=request_body.encode("utf-8"), timeout=30)
    print(f"HTTP {resp.status_code}: {resp.text}")
    if resp.status_code != 200:
        sys.exit(1)

if __name__ == "__main__":
    main()

DLC ジョブ送信時のタグ指定

CreateJob API を使用して DLC ジョブを送信する際は、Settings.Tags フィールドにタグを指定します。すべての必須タグを入力する必要があります。カスタム入力が可能な ValueFlexible が true のタグを除き、値は事前定義されたオプションから選択する必要があります。

Python コード例：DLC ジョブ送信時のタグ指定

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
PAI-DLC トレーニングジョブの送信 (事前定義されたタグ付き)
API: POST /api/v1/jobs (API バージョン 2020-12-03)
タグは Settings.Tags に辞書 {"tag-name": "tag-value"} として配置されます。
"""

import json, sys, hashlib, hmac, datetime, time, requests

ACCESS_KEY_ID     = "your-AccessKeyId"
ACCESS_KEY_SECRET = "your-AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "your-workspace-ID"
RESOURCE_ID       = ""  # リソースクォータ ID。空のままにするとパブリックリソースが使用されます。

DLC_ENDPOINT = f"pai-dlc.{REGION_ID}.aliyuncs.com"
DLC_API_PATH = "/api/v1/jobs"

# ワークスペースの事前定義されたタグに従って入力します。すべての事前定義されたタグを指定する必要があります。
TAGS = {
    "Scene":        "Training",
    "ModelType":    "NLP-Text",
    "ExpDirection": "Fine-grained-DataProcessing",
}

def sign_request(method, path, headers, body, action="CreateJob"):
    now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    headers.update({"x-acs-date": now, "x-acs-version": "2020-12-03", "x-acs-action": action,
                    "x-acs-signature-nonce": hashlib.md5((now + body).encode()).hexdigest()})
    signed_keys = sorted(k for k in headers if k.startswith("x-acs-") or k in ("host", "content-type"))
    canonical_headers = "".join(f"{k}:{headers[k]}\n" for k in signed_keys)
    hashed_body = hashlib.sha256(body.encode("utf-8")).hexdigest()
    canonical_request = f"{method}\n{path}\n\n{canonical_headers}\n{';'.join(signed_keys)}\n{hashed_body}"
    string_to_sign = f"ACS3-HMAC-SHA256\n{hashlib.sha256(canonical_request.encode()).hexdigest()}"
    signature = hmac.new(ACCESS_KEY_SECRET.encode(), string_to_sign.encode(), hashlib.sha256).hexdigest()
    headers["Authorization"] = (f"ACS3-HMAC-SHA256 Credential={ACCESS_KEY_ID},"
                                 f"SignedHeaders={';'.join(signed_keys)},Signature={signature}")
    return headers

def main():
    job = {
        "DisplayName": f"my-training-job-{int(time.time())}",
        "JobType": "PyTorchJob",
        "WorkspaceId": WORKSPACE_ID,
        "UserCommand": "echo hello && sleep 5",
        "JobSpecs": [{"Type": "Worker",
                      "Image": "registry.cn-hangzhou.aliyuncs.com/pai-dlc/pytorch-training:1.12-gpu-py39-cu116-ubuntu20.04",
                      "PodCount": 1, "EcsSpec": "ecs.g6.large"}],
        "Settings": {"Tags": TAGS},  # タグは Settings.Tags に配置されます
    }
    if RESOURCE_ID:
        job["ResourceId"] = RESOURCE_ID
    request_body = json.dumps(job, ensure_ascii=False)
    url = f"https://{DLC_ENDPOINT}{DLC_API_PATH}"
    headers = sign_request("POST", DLC_API_PATH, {"host": DLC_ENDPOINT, "content-type": "application/json"}, request_body)
    resp = requests.post(url, headers=headers, data=request_body.encode("utf-8"), timeout=30)
    print(f"HTTP {resp.status_code}: {resp.text}")
    if resp.status_code != 200:
        sys.exit(1)

if __name__ == "__main__":
    main()

タグによる DLC ジョブのフィルタリング

ListJobs API は、Tags クエリパラメーター (JSON 文字列形式) を使用したタグベースのフィルタリングをサポートしています。複数のタグは AND ロジックを使用します。つまり、ジョブが結果に表示されるには、指定されたすべてのタグが一致する必要があります。

Python コード例：タグによる DLC ジョブのフィルタリング

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# PAI-DLC ジョブをタグでフィルタリングしてリスト表示
# API: GET /api/v1/jobs (API バージョン 2020-12-03)
# 使用する前に、ACCESS_KEY_ID、ACCESS_KEY_SECRET、REGION_ID、および WORKSPACE_ID を入力してください。

import json
import sys
import hashlib
import hmac
import datetime
import requests
from urllib.parse import urlencode

ACCESS_KEY_ID     = "your-AccessKeyId"
ACCESS_KEY_SECRET = "your-AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "your-workspace-ID"

ENDPOINT = f"pai-dlc.{REGION_ID}.aliyuncs.com"
API_PATH = "/api/v1/jobs"

def build_query_params(workspace_id, filter_tags, page_size=50, page_number=1):
    query_params = {
        "WorkspaceId": workspace_id,
        "PageSize": str(page_size),
        "PageNumber": str(page_number),
        "SortBy": "GmtCreateTime",
        "Order": "desc",
    }

    if filter_tags:
        query_params["Tags"] = json.dumps(
            filter_tags, ensure_ascii=False, separators=(",", ":")
        )

    return query_params

def sign_request(method, path, query_string, headers, body=""):
    now = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ")
    headers.update({
        "x-acs-date": now,
        "x-acs-version": "2020-12-03",
        "x-acs-action": "ListJobs",
        "x-acs-signature-nonce": hashlib.md5((now + query_string + body).encode()).hexdigest(),
    })

    normalized_headers = {k.lower(): v for k, v in headers.items()}
    signed_keys = sorted(
        k for k in normalized_headers
        if k.startswith("x-acs-") or k in ("host", "content-type")
    )

    canonical_headers = "".join(f"{k}:{normalized_headers[k]}\n" for k in signed_keys)
    hashed_body = hashlib.sha256(body.encode("utf-8")).hexdigest()

    canonical_request = (
        f"{method}\n"
        f"{path}\n"
        f"{query_string}\n"
        f"{canonical_headers}\n"
        f"{';'.join(signed_keys)}\n"
        f"{hashed_body}"
    )

    string_to_sign = f"ACS3-HMAC-SHA256\n{hashlib.sha256(canonical_request.encode()).hexdigest()}"
    signature = hmac.new(
        ACCESS_KEY_SECRET.encode(),
        string_to_sign.encode(),
        hashlib.sha256
    ).hexdigest()

    headers["Authorization"] = (
        f"ACS3-HMAC-SHA256 Credential={ACCESS_KEY_ID},"
        f"SignedHeaders={';'.join(signed_keys)},"
        f"Signature={signature}"
    )
    return headers

def list_jobs_by_tags(filter_tags, page_size=50, page_number=1):
    query_params = build_query_params(WORKSPACE_ID, filter_tags, page_size, page_number)
    query_string = urlencode(sorted(query_params.items()))
    url = f"https://{ENDPOINT}{API_PATH}?{query_string}"

    headers = {
        "host": ENDPOINT,
        "content-type": "application/json",
    }
    headers = sign_request("GET", API_PATH, query_string, headers, body="")

    resp = requests.get(url, headers=headers, timeout=30)
    print(f"HTTP {resp.status_code}")
    print(resp.text)

    if resp.status_code != 200:
        sys.exit(1)

    return resp.json()

def main():
    filter_tags = {"Scene": "Training"}
    list_jobs_by_tags(filter_tags)

if __name__ == "__main__":
    main()

タグ変更履歴の表示

ActionTrail は、操作者、タイムスタンプ、変更内容など、タグルールへのすべての変更を記録します。

ActionTrail コンソールにログインし、[Advanced query] に移動し、サービス名を [PAI]、イベント名を [UpdateConfigs] に設定し、ConfigKey が [predefinedTags] のレコードをフィルタリングして、タグルールの変更履歴を表示します。

タグ ルールの定義