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] をクリックします。
ジョブ設定ページでタグセクションを見つけ、ワークスペースのタグ規則に従ってキーと値のペアを入力します。
ジョブ作成時には、スクリーンショットにある「self-define-key」のようなカスタムタグも追加できます。
必要なタグをすべて入力した後、ジョブを送信します。

重要

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

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

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

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

付録

API リファレンス

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

タグ規則の一括設定

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

リクエストボディのサンプル構造：

{
  "Configs": [
    {
      "ConfigKey": "predefinedTags",
      "ConfigValue": "[{\"Type\":\"Tag\",\"Key\":\"scenario\",\"Value\":\"{\\\"Products\\\":\\\"DLC,DSW,EAS\\\",\\\"Values\\\":\\\"Training,Evaluation,Inference\\\",\\\"ValueFlexible\\\":false,\\\"Description\\\":\\\"シナリオを選択してください\\\"}\"}]",
      "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     = "ご自身の AccessKeyId"
ACCESS_KEY_SECRET = "ご自身の AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "ご自身のワークスペース 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     = "ご自身の AccessKeyId"
ACCESS_KEY_SECRET = "ご自身の AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "ご自身のワークスペース 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 -*-

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

ACCESS_KEY_ID     = "ご自身の AccessKeyId"
ACCESS_KEY_SECRET = "ご自身の AccessKeySecret"
REGION_ID         = "cn-hangzhou"
WORKSPACE_ID      = "ご自身のワークスペース 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] であるレコードでフィルタリングして、タグ規則の変更履歴を表示します。