创建Workflow训练作业节点

功能介绍

该节点通过对算法、输入、输出的定义，实现ModelArts作业管理的能力。主要用于数据处理、模型训练、模型评估等场景。主要应用场景如下：

• 当需要对图像进行增强，对语音进行除噪等操作时，可以使用该节点进行数据的预处理。
• 对于一些物体检测，图像分类等模型场景，可以根据已有的数据使用该节点进行模型的训练。

属性总览

您可以使用JobStep来构建作业类型节点，JobStep结构如下

资源规格查询

您在创建作业类型节点之前可以通过以下操作来获取该账号所支持的训练资源规格列表以及引擎规格列表：

• 导包

  from modelarts.session import Session
  from modelarts.estimatorV2 import TrainingJob
  from modelarts.workflow.client.job_client import JobClient

• session初始化

  # 如果您在本地IDEA环境中开发工作流，则Session初始化使用如下方式
  # 认证用的ak和sk硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全；
  # 本示例以ak和sk保存在环境变量中来实现身份验证为例，运行本示例前请先在本地环境中设置环境变量HUAWEICLOUD_SDK_AK和HUAWEICLOUD_SDK_SK。
  __AK = os.environ["HUAWEICLOUD_SDK_AK"]
  __SK = os.environ["HUAWEICLOUD_SDK_SK"]
  # 如果进行了加密还需要进行解密操作
  session = Session(
      access_key=__AK, # 账号的AK信息
      secret_key=__SK, # 账号的SK信息
      region_name="***", # 账号所属的region
      project_id="***" ,# 账号的项目ID
  )
  
  # 如果您在Notebook环境中开发工作流，则Session初始化使用如下方式
  session = Session()

• 公共池查询

  # 公共资源池规格列表查询
  spec_list = TrainingJob(session).get_train_instance_types(session) # 返回的类型为list,可按需打印查看
  print(spec_list)

• 专属池查询

  # 运行中的专属资源池列表查询
  pool_list = JobClient(session).get_pool_list() # 返回专属资源池的详情列表
  pool_id_list = JobClient(session).get_pool_id_list() # 返回专属资源池ID列表
  专属资源池规格ID列表如下，根据所选资源池的实际规格自行选择：
      1. modelarts.pool.visual.xlarge 对应1卡
      2. modelarts.pool.visual.2xlarge 对应2卡
      3. modelarts.pool.visual.4xlarge 对应4卡
      4. modelarts.pool.visual.8xlarge 对应8卡

• 引擎规格查询

  # 引擎规格查询
  engine_dict = TrainingJob(session).get_engine_list(session) # 返回的类型为dict,可按需打印查看
  print(engine_dict)

使用案例

主要包含七种场景的用例：

• 使用订阅自AI Gallery的算法
• 使用算法管理中的算法
• 使用自定义算法（代码目录+启动文件+官方镜像）
• 使用自定义算法（代码目录+脚本命令+自定义镜像）
• 基于数据集版本发布节点构建作业类型节点
• 作业类型节点结合可视化能力
• 输入使用DataSelector对象，支持选择OBS或者数据集

使用订阅自AI Gallery的算法

from modelarts import workflow as wf

# 构建一个OutputStorage对象，对训练输出目录做统一管理
storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 

# 定义输入的数据集对象
dataset = wf.data.DatasetPlaceholder(name="input_dataset")

# 通过JobStep来定义一个训练节点，输入使用数据集，并将训练结果输出到OBS
job_step = wf.steps.JobStep(
    name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
    title="图像分类训练", # 标题信息，不填默认使用name
    algorithm=wf.AIGalleryAlgorithm(
        subscription_id="subscription_id", # 算法订阅ID，也可直接填写版本号
        item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号
        parameters=[
            wf.AlgorithmParameters(
                name="parameter_name", 
                value=wf.Placeholder(name="parameter_name", placeholder_type=wf.PlaceholderType.STR, default="fake_value",description="description_info")
            ) # 算法超参的值使用Placeholder对象来表示，支持int, bool, float, str四种类型
        ]
    ),  # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值
    
    inputs=wf.steps.JobInput(name="data_url", data=dataset), # JobStep的输入在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="fake_dataset_name", version_name="fake_version_name")表示
    outputs=wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))), # JobStep的输出
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格")
           
        )
    )# 训练资源规格信息
)

workflow = wf.Workflow(
    name="job-step-demo",
    desc="this is a demo workflow",
    steps=[job_step],
    storages=[storage]
)

使用算法管理中的算法

from modelarts import workflow as wf

# 构建一个OutputStorage对象，对训练输出目录做统一管理
storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 

# 定义输入的数据集对象
dataset = wf.data.DatasetPlaceholder(name="input_dataset")

# 通过JobStep来定义一个训练节点，输入使用数据集，并将训练结果输出到OBS
job_step = wf.steps.JobStep(
    name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
    title="图像分类训练", # 标题信息，不填默认使用name
    algorithm=wf.Algorithm(
        algorithm_id="algorithm_id", # 算法ID
        parameters=[
            wf.AlgorithmParameters(
                name="parameter_name", 
                value=wf.Placeholder(name="parameter_name", placeholder_type=wf.PlaceholderType.STR, default="fake_value",description="description_info")
            ) # 算法超参的值使用Placeholder对象来表示，支持int, bool, float, str四种类型
        ]
    ), # 训练使用的算法对象，示例中的算法来源于算法管理；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值

    inputs=wf.steps.JobInput(name="data_url", data=dataset), # JobStep的输入在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="fake_dataset_name", version_name="fake_version_name")表示
    outputs=wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))), # JobStep的输出
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格")

        )
    )# 训练资源规格信息
)

workflow = wf.Workflow(
    name="job-step-demo",
    desc="this is a demo workflow",
    steps=[job_step],
    storages=[storage]
)

使用自定义算法（代码目录+启动文件+官方镜像）

from modelarts import workflow as wf

# 构建一个OutputStorage对象，对训练输出目录做统一管理
storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 

# 定义输入的数据集对象
dataset = wf.data.DatasetPlaceholder(name="input_dataset")

# 通过JobStep来定义一个训练节点，输入使用数据集，并将训练结果输出到OBS
job_step = wf.steps.JobStep(
    name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
    title="图像分类训练", # 标题信息，不填默认使用name
    algorithm=wf.BaseAlgorithm(
        code_dir="fake_code_dir", # 代码目录存储的路径
        boot_file="fake_boot_file", # 启动文件存储路径，需要在代码目录下
        engine=wf.steps.JobEngine(engine_name="fake_engine_name", engine_version="fake_engine_version"), # 官方镜像的名称以及版本信息

        parameters=[
            wf.AlgorithmParameters(
                name="parameter_name", 
                value=wf.Placeholder(name="parameter_name", placeholder_type=wf.PlaceholderType.STR, default="fake_value",description="description_info")
            ) # 算法超参的值使用Placeholder对象来表示，支持int, bool, float, str四种类型
        ]
    ), # 自定义算法使用代码目录+启动文件+官方镜像的方式实现

    
    inputs=wf.steps.JobInput(name="data_url", data=dataset), # JobStep的输入在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="fake_dataset_name", version_name="fake_version_name")表示
    outputs=wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))), # JobStep的输出
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格")
            
        )
    )# 训练资源规格信息
)

workflow = wf.Workflow(
    name="job-step-demo",
    desc="this is a demo workflow",
    steps=[job_step],
    storages=[storage]
)

使用自定义算法（代码目录+脚本命令+自定义镜像）

from modelarts import workflow as wf

# 构建一个OutputStorage对象，对训练输出目录做统一管理
storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 

# 定义输入的数据集对象
dataset = wf.data.DatasetPlaceholder(name="input_dataset")

# 通过JobStep来定义一个训练节点，输入使用数据集，并将训练结果输出到OBS
job_step = wf.steps.JobStep(
    name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
    title="图像分类训练", # 标题信息，不填默认使用name
    algorithm=wf.BaseAlgorithm(
        code_dir="fake_code_dir", # 代码目录存储的路径
        command="fake_command", # 执行的脚本命令
        engine=wf.steps.JobEngine(image_url="fake_image_url"), # 自定义镜像的url，格式为：组织名/镜像名称:版本号，不需要携带相应的域名地址；如果image_url需要设置为运行态可配置，则使用如下方式：image_url=wf.Placeholder(name="image_url", placeholder_type=wf.PlaceholderType.STR, placeholder_format="swr", description="自定义镜像")
        parameters=[
            wf.AlgorithmParameters(
                name="parameter_name", 
                value=wf.Placeholder(name="parameter_name", placeholder_type=wf.PlaceholderType.STR, default="fake_value",description="description_info")
            ) # 算法超参的值使用Placeholder对象来表示，支持int, bool, float, str四种类型
        ]
    ), 自定义算法使用代码目录+脚本命令+自定义镜像的方式实现

    inputs=wf.steps.JobInput(name="data_url", data=dataset), # JobStep的输入在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="fake_dataset_name", version_name="fake_version_name")表示
    outputs=wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))), # JobStep的输出
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格")
            
        )
    )# 训练资源规格信息
)

workflow = wf.Workflow(
    name="job-step-demo",
    desc="this is a demo workflow",
    steps=[job_step],
    storages=[storage]
)

上述四种方式使用据集对象作为输入，如果您需要使用OBS路径作为输入时，只需将JobInput中的data数据替换为data=wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory")或者data=wf.data.OBSPath(obs_path="fake_obs_path")即可。

此外，在构建工作流时就指定好数据集对象或者OBS路径的方式可以减少配置操作，方便您在开发态进行调试。但是对于发布到运行态或者gallery的工作流，更推荐的方式是采用数据占位符的方式进行编写，您可以在工作流启动之前对参数进行配置，自由度更高。

基于数据集版本发布节点构建作业类型节点

使用场景：数据集版本发布节点的输出作为作业类型节点的输入。

from modelarts import workflow as wf

# 定义数据集对象
dataset = wf.data.DatasetPlaceholder(name="input_dataset")

# 定义训练验证切分比参数
train_ration = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR, default="0.8")

release_version_step = wf.steps.ReleaseDatasetStep(
    name="release_dataset", # 数据集发布节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
    title="数据集版本发布", # 标题信息，不填默认使用name值
    inputs=wf.steps.ReleaseDatasetInput(name="input_name", data=dataset), # ReleaseDatasetStep的输入，数据集对象在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="dataset_name")表示
    outputs=wf.steps.ReleaseDatasetOutput(
        name="output_name", 
        dataset_version_config=wf.data.DatasetVersionConfig(
            label_task_type=wf.data.LabelTaskTypeEnum.IMAGE_CLASSIFICATION,  # 数据集发布版本时需要指定标注任务的类型
            train_evaluate_sample_ratio=train_ration # 数据集的训练验证切分比
            )
    ) # ReleaseDatasetStep的输出
)

# 构建一个OutputStorage对象，对训练输出目录做统一管理
storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 

# 通过JobStep来定义一个训练节点，输入使用数据集，并将训练结果输出到OBS
job_step = wf.steps.JobStep(
    name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
    title="图像分类训练", # 标题信息，不填默认使用name
    algorithm=wf.AIGalleryAlgorithm(
        subscription_id="subscription_id", # 算法订阅ID
        item_version_id="item_version_id", # 算法订阅版本ID
        parameters=[
            wf.AlgorithmParameters(
                name="parameter_name", 
                value=wf.Placeholder(name="parameter_name", placeholder_type=wf.PlaceholderType.STR, default="fake_value",description="description_info")
            ) # 算法超参的值使用Placeholder对象来表示，支持int, bool, float, str四种类型
        ]
    ),  # 训练使用的算法对象，示例中使用AI Gallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值

    
    inputs=wf.steps.JobInput(name="data_url", data=release_version_step.outputs["output_name"].as_input()), # 使用数据集版本发布节点的输出作为JobStep的输入
    outputs=wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))), # JobStep的输出
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格")
            
        )
    ), # 训练资源规格信息
    depend_steps=release_version_step # 依赖的数据集版本发布节点对象
)
# release_version_step是wf.steps.ReleaseDatasetStep的实例对象，output_name是wf.steps.ReleaseDatasetOutput的name字段值

workflow = wf.Workflow(
    name="job-step-demo",
    desc="this is a demo workflow",
    steps=[release_version_step, job_step],
    storages=[storage]
)

作业类型节点结合可视化能力

节点可视化特性将用户在使用Workflow时产生的一些衡量指标进行一个可视化的展示，支持数据的实时可视化，并且允许独立呈现可视化外挂节点。形态上基于作业类型节点原有的使用方式，新增一个针对metrics信息展示的输出，通过MetricsConfig对象进行配置。

对于输出的metrics文件，数据内容必须为标准的json数据，大小限制为1M，并且与当前支持的几种数据格式保持一致：

• 键值对类型的数据

  [
      {
          "key": "loss",
          "title": "loss",
          "type": "float",
          "data": {
              "value": 1.2
          }
      },
      {
          "key": "accuracy",
          "title": "accuracy",
          "type": "float",
          "data": {
              "value": 1.6
          }
      }
  ]

• 折线图数据(type是line chart)

  [
      {
          "key": "metric",
          "title": "metric",
          "type": "line chart",
          "data": {
              "x_axis": [
                  {
                      "title": "step/epoch",
                      "value": [
                          1,
                          2,
                          3
                      ]
                  }
              ],
              "y_axis": [
                  {
                      "title": "value",
                      "value": [
                          0.5,
                          0.4,
                          0.3
                      ]
                  }
              ]
          }
      }
  ]

• 柱状图数据(type是histogram)

  [
      {
          "key": "metric",
          "title": "metric",
          "type": "histogram",
          "data": {
              "x_axis": [
                  {
                      "title": "step/epoch",
                      "value": [
                          1,
                          2,
                          3
                      ]
                  }
              ],
              "y_axis": [
                  {
                      "title": "value",
                      "value": [
                          0.5,
                          0.4,
                          0.3
                      ]
                  }
              ]
          }
      }
  ]

• 混淆矩阵

  [
      {
          "key": "confusion_matrix",
          "title": "confusion_matrix",
          "type": "table",
          "data": {
              "cell_value": [
                  [
                      1,
                      2
                  ],
                  [
                      2,
                      3
                  ]
              ],
              "col_labels": {
                  "title": "labels",
                  "value": [
                      "daisy",
                      "dandelion"
                  ]
              },
              "row_labels": {
                  "title": "predictions",
                  "value": [
                      "daisy",
                      "dandelion"
                  ]
              }
          }
      }
  ]

• 一维表格

  [
      {
          "key": "Application Evaluation Results",
          "title": "Application Evaluation Results",
          "type": "one-dimensional-table",
          "data": {
              "cell_value": [
                  [
                      10,
                      2,
                      0.5
                  ]
              ],
              "labels": [
                  "samples",
                  "maxResTine",
                  "p99"
              ]
          }
      }
  ]

  使用案例：

  from modelarts import workflow as wf
  
  # 构建一个Storage对象，对训练输出目录做统一管理
  storage = wf.data.Storage(name="storage_name", title="title_info", description="description_info", with_execution_id=True, create_dir=True) # name字段必填，title, description可选填 
  
  # 定义输入的数据集对象
  dataset = wf.data.DatasetPlaceholder(name="input_dataset")
  
  # 通过JobStep来定义一个训练节点，输入使用数据集，并将训练结果输出到OBS
  job_step = wf.steps.JobStep(
      name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
      title="图像分类训练", # 标题信息，不填默认使用name
      algorithm=wf.AIGalleryAlgorithm(
          subscription_id="subscription_id", # 算法订阅ID
          item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号
          parameters=[
              wf.AlgorithmParameters(
                  name="parameter_name", 
                  value=wf.Placeholder(name="parameter_name", placeholder_type=wf.PlaceholderType.STR, default="fake_value",description="description_info")
              ) # 算法超参的值使用Placeholder对象来表示，支持int, bool, float, str四种类型
          ]
  
      ),  # 训练使用的算法对象，示例中使用AI Gallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值
  
      
      inputs=wf.steps.JobInput(name="data_url", data=dataset), # JobStep的输入在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="fake_dataset_name", version_name="fake_version_name")表示
      outputs=[
          wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))),# JobStep的输出
          wf.steps.JobOutput(name="metrics_output", metrics_config=wf.data.MetricsConfig(metric_files=storage.join("directory_path/metrics.json", create_dir=False))) # 相关metrics信息由作业的脚本代码自行输出到配置的路径下
      ], 
      spec=wf.steps.JobSpec(
          resource=wf.steps.JobResource(
              flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格")
              
          )
      )# 训练资源规格信息
  )
  
  workflow = wf.Workflow(
      name="job-step-demo",
      desc="this is a demo workflow",
      steps=[job_step],
      storages=[storage]
  )

Workflow不会自动获取训练输出的指标信息，要求用户自行在算法代码中获取指标信息并且按照指定的数据格式构造出metrics.json文件，自行上传到MetricsConfig中配置的OBS路径下，Workflow只进行数据的读取以及渲染展示。

输入使用DataSelector对象，支持选择OBS或者数据集

该方式主要用于输入支持可选择的场景，使用DataSelector对象作为输入时，用户在页面配置时可自由选择数据集对象或者OBS对象作为训练的输入，代码示例如下：

from modelarts import workflow as wf

# 构建一个OutputStorage对象，对训练输出目录做统一管理
storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 

# 定义DataSelector对象
data_selector = wf.data.DataSelector(name="input_data", data_type_list=["dataset", "obs"])

# 通过JobStep来定义一个训练节点，输入使用数据集，并将训练结果输出到OBS
job_step = wf.steps.JobStep(
    name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复
    title="图像分类训练", # 标题信息，不填默认使用name
    algorithm=wf.AIGalleryAlgorithm(
        subscription_id="subscription_id", # 算法订阅ID，也可直接填写版本号
        item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号
        parameters=[
            wf.AlgorithmParameters(
                name="parameter_name", 
                value=wf.Placeholder(name="parameter_name", placeholder_type=wf.PlaceholderType.STR, default="fake_value",description="description_info")
            ) # 算法超参的值使用Placeholder对象来表示，支持int, bool, float, str四种类型
        ]
    ),  # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值
    
    inputs=wf.steps.JobInput(name="data_url", data=data_selector), # JobStep的输入在运行时配置,可自由选择OBS或者数据集作为输入
    outputs=wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))), # JobStep的输出
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格")
           
        )
    )# 训练资源规格信息
)

workflow = wf.Workflow(
    name="job-step-demo",
    desc="this is a demo workflow",
    steps=[job_step],
    storages=[storage]
)

使用DataSelector作为输入时，需要用户自行保证算法的输入同时支持数据集或者OBS。