功能介紹

支持單節(jié)點(diǎn)通過參數(shù)配置或者獲取訓(xùn)練輸出的metric指標(biāo)信息來決定執(zhí)行是否跳過，同時(shí)可以基于此能力完成對(duì)執(zhí)行流程的控制。

應(yīng)用場景

主要用于存在多分支選擇執(zhí)行的復(fù)雜場景，在每次啟動(dòng)執(zhí)行后需要根據(jù)相關(guān)配置信息決定哪些分支需要執(zhí)行，哪些分支需要跳過，達(dá)到分支部分執(zhí)行的目的，與ConditionStep的使用場景類似，但功能更加強(qiáng)大。當(dāng)前該能力適用于數(shù)據(jù)集創(chuàng)建節(jié)點(diǎn)、數(shù)據(jù)集標(biāo)注節(jié)點(diǎn)、數(shù)據(jù)集導(dǎo)入節(jié)點(diǎn)、數(shù)據(jù)集版本發(fā)布節(jié)點(diǎn)、作業(yè)類型節(jié)點(diǎn)、模型注冊(cè)節(jié)點(diǎn)以及服務(wù)部署節(jié)點(diǎn)。

控制單節(jié)點(diǎn)的執(zhí)行

• 通過參數(shù)配置實(shí)現(xiàn)

  from modelarts import workflow as wf
  
  condition_equal = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.EQ, left=wf.Placeholder(name="is_skip", placeholder_type=wf.PlaceholderType.BOOL), right=True)
  
  # 構(gòu)建一個(gè)OutputStorage對(duì)象，對(duì)訓(xùn)練輸出目錄做統(tǒng)一管理
  storage = wf.data.OutputStorage(name="storage_name", title="title_info",
                                  description="description_info")  # name字段必填，title, description可選填
  
  # 定義輸入的 OBS 對(duì)象
  obs_data = wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory")
  
  # 通過JobStep來定義一個(gè)訓(xùn)練節(jié)點(diǎn)，并將訓(xùn)練結(jié)果輸出到OBS
  job_step = wf.steps.JobStep(
      name="training_job",  # 訓(xùn)練節(jié)點(diǎn)的名稱，命名規(guī)范(只能包含英文字母、數(shù)字、下劃線（_）、中劃線（-），并且只能以英文字母開頭，長度限制為64字符)，一個(gè)Workflow里的兩個(gè)step名稱不能重復(fù)
      title="圖像分類訓(xùn)練",  # 標(biāo)題信息，不填默認(rèn)使用name
      algorithm=wf.AIGalleryAlgorithm(
          subscription_id="subscription_id",  # 算法訂閱ID
          item_version_id="item_version_id",  # 算法訂閱版本ID，也可直接填寫版本號(hào)
          parameters=[]
  
      ),  # 訓(xùn)練使用的算法對(duì)象，示例中使用AIGallery訂閱的算法；部分算法超參的值如果無需修改，則在parameters字段中可以不填寫，系統(tǒng)自動(dòng)填充相關(guān)超參值
  
      inputs=wf.steps.JobInput(name="data_url", data=obs_data),
      # JobStep的輸入在運(yùn)行時(shí)配置；data字段也可使用data=wf.data.OBSPath(obs_path="fake_obs_path")表示
      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="訓(xùn)練資源規(guī)格")
  
          )
      ), # 訓(xùn)練資源規(guī)格信息
      policy=wf.steps.StepPolicy(
          skip_conditions=[condition_equal] # 通過skip_conditions中的計(jì)算結(jié)果決定job_step是否跳過
      )
  )
  
  workflow = wf.Workflow(
      name="new-condition-demo",
      desc="this is a demo workflow",
      steps=[job_step],
      storages=storage
  )

  案例中job_step配置了相關(guān)的跳過策略，并且通過一個(gè)bool類型的參數(shù)進(jìn)行控制。當(dāng)name為is_skip的Placeholder參數(shù)配置為True時(shí)，condition_equal的計(jì)算結(jié)果為True，此時(shí)job_step會(huì)被置為跳過，反之job_step正常執(zhí)行，其中Condition對(duì)象詳情可參考構(gòu)建條件節(jié)點(diǎn)控制分支執(zhí)行。

• 通過獲取JobStep輸出的相關(guān)metric指標(biāo)信息實(shí)現(xiàn)

  from modelarts import workflow as wf
  
  # 構(gòu)建一個(gè)OutputStorage對(duì)象，對(duì)訓(xùn)練輸出目錄做統(tǒng)一管理
  storage = wf.data.Storage(name="storage_name", title="title_info", with_execution_id=True, create_dir=True, description="description_info")  # name字段必填，title, description可選填
  
  # 定義輸入的OBS對(duì)象
  obs_data = wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory")
  
  # 通過JobStep來定義一個(gè)訓(xùn)練節(jié)點(diǎn)，并將訓(xùn)練結(jié)果輸出到OBS
  job_step = wf.steps.JobStep(
      name="training_job",  # 訓(xùn)練節(jié)點(diǎn)的名稱，命名規(guī)范(只能包含英文字母、數(shù)字、下劃線（_）、中劃線（-），并且只能以英文字母開頭，長度限制為64字符)，一個(gè)Workflow里的兩個(gè)step名稱不能重復(fù)
      title="圖像分類訓(xùn)練",  # 標(biāo)題信息，不填默認(rèn)使用name
      algorithm=wf.AIGalleryAlgorithm(
          subscription_id="subscription_id",  # 算法訂閱ID
          item_version_id="item_version_id",  # 算法訂閱版本ID，也可直接填寫版本號(hào)
          parameters=[]
  
      ),  # 訓(xùn)練使用的算法對(duì)象，示例中使用AIGallery訂閱的算法；部分算法超參的值如果無需修改，則在parameters字段中可以不填寫，系統(tǒng)自動(dòng)填充相關(guān)超參值
      inputs=wf.steps.JobInput(name="data_url", data=obs_data),
      outputs=[
          wf.steps.JobOutput(name="train_url",obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path"))),
          wf.steps.JobOutput(name="metrics", metrics_config=wf.data.MetricsConfig(metric_files=storage.join("directory_path/metrics.json", create_dir=False))) # 指定metric的輸出路徑，相關(guān)指標(biāo)信息由作業(yè)腳本代碼根據(jù)指定的數(shù)據(jù)格式自行輸出（示例中需要將metric信息輸出到訓(xùn)練輸出目錄下的metrics.json文件中）
      ],
      spec=wf.steps.JobSpec(
          resource=wf.steps.JobResource(
              flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="訓(xùn)練資源規(guī)格")
          )
      )  # 訓(xùn)練資源規(guī)格信息
  )
  
  # 定義模型名稱參數(shù)
  model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR)
  
  # 定義條件對(duì)象
  condition_lt = wf.steps.Condition(
      condition_type=wf.steps.ConditionTypeEnum.LT,
      left=wf.steps.MetricInfo(job_step.outputs["metrics"].as_input(), "accuracy"),
      right=0.5
  )
  
  model_step = wf.steps.ModelStep(
      name="model_registration",  # 模型注冊(cè)節(jié)點(diǎn)的名稱，命名規(guī)范(只能包含英文字母、數(shù)字、下劃線（_）、中劃線（-），并且只能以英文字母開頭，長度限制為64字符)，一個(gè)Workflow里的兩個(gè)step名稱不能重復(fù)
      title="模型注冊(cè)",  # 標(biāo)題信息
      inputs=wf.steps.ModelInput(name='model_input', data=job_step.outputs["train_url"].as_input()),  # job_step的輸出作為輸入
      outputs=wf.steps.ModelOutput(name='model_output', model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow")),  # ModelStep的輸出
      depend_steps=[job_step],  # 依賴的作業(yè)類型節(jié)點(diǎn)對(duì)象
      policy=wf.steps.StepPolicy(skip_conditions=condition_lt) # 通過skip_conditions中的計(jì)算結(jié)果決定model_step是否跳過
  )
  
  workflow = wf.Workflow(
      name="new-condition-demo",
      desc="this is a demo workflow",
      steps=[job_step, model_step],
      storages=storage
  )

  案例中model_step配置了相關(guān)的跳過策略，并且通過獲取job_step輸出的accuracy指標(biāo)信息與預(yù)置的值進(jìn)行比較，決定是否需要進(jìn)行模型注冊(cè)。當(dāng)job_step輸出的accuracy指標(biāo)數(shù)據(jù)小于閾值0.5時(shí)，condition_lt的計(jì)算結(jié)果為True，此時(shí)model_step會(huì)被置為跳過，反之model_step正常執(zhí)行。

  

  job_step輸出的metric文件格式要求可參考創(chuàng)建Workflow訓(xùn)練作業(yè)節(jié)點(diǎn)部分，并且在Condition中只支持使用type為float類型的指標(biāo)數(shù)據(jù)作為輸入。

  此案例中metrics.json的內(nèi)容示例如下：

  [
      {
          "key": "loss", // 指標(biāo)數(shù)據(jù)名稱，不支持特殊字符，長度限制為64字符
          "title": "loss", // 指標(biāo)數(shù)據(jù)標(biāo)題，長度限制為64字符
          "type": "float", // 指標(biāo)數(shù)據(jù)類型，支持以下類型：浮點(diǎn)：float、折線圖：line chart、柱狀圖：histogram、矩陣：table、一維表格：one-dimensional-table
          "data": {
              "value": 1.2 // 指標(biāo)數(shù)據(jù)值，不同類型的使用示例可以參考創(chuàng)建Workflow訓(xùn)練作業(yè)節(jié)點(diǎn)
          }
      },
      {
          "key": "accuracy",
          "title": "accuracy",
          "type": "float",
          "data": {
              "value": 0.8
          }
      }   
  ]

控制多分支的部分執(zhí)行

from modelarts import workflow as wf

# 構(gòu)建一個(gè)OutputStorage對(duì)象，對(duì)訓(xùn)練輸出目錄做統(tǒng)一管理
storage = wf.data.Storage(name="storage_name", title="title_info", with_execution_id=True, create_dir=True, description="description_info")  # name字段必填，title, description可選填

# 定義輸入的OBS對(duì)象
obs_data = wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory")

condition_equal_a = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.EQ, left=wf.Placeholder(name="job_step_a_is_skip", placeholder_type=wf.PlaceholderType.BOOL), right=True)

# 通過JobStep來定義一個(gè)訓(xùn)練節(jié)點(diǎn)，并將訓(xùn)練結(jié)果輸出到OBS
job_step_a = wf.steps.JobStep(
    name="training_job_a",  # 訓(xùn)練節(jié)點(diǎn)的名稱，命名規(guī)范(只能包含英文字母、數(shù)字、下劃線（_）、中劃線（-），并且只能以英文字母開頭，長度限制為64字符)，一個(gè)Workflow里的兩個(gè)step名稱不能重復(fù)
    title="圖像分類訓(xùn)練",  # 標(biāo)題信息，不填默認(rèn)使用name
    algorithm=wf.AIGalleryAlgorithm(
        subscription_id="subscription_id",  # 算法訂閱ID
        item_version_id="item_version_id",  # 算法訂閱版本ID，也可直接填寫版本號(hào)
        parameters=[]

    ),  # 訓(xùn)練使用的算法對(duì)象，示例中使用AIGallery訂閱的算法；部分算法超參的值如果無需修改，則在parameters字段中可以不填寫，系統(tǒng)自動(dòng)填充相關(guān)超參值
    inputs=wf.steps.JobInput(name="data_url", data=obs_data),
    outputs=[wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path_a")))],
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="訓(xùn)練資源規(guī)格")

        )
    ),  # 訓(xùn)練資源規(guī)格信息
    policy=wf.steps.StepPolicy(skip_conditions=condition_equal_a)
)

condition_equal_b = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.EQ, left=wf.Placeholder(name="job_step_b_is_skip", placeholder_type=wf.PlaceholderType.BOOL), right=True)

# 通過JobStep來定義一個(gè)訓(xùn)練節(jié)點(diǎn)，并將訓(xùn)練結(jié)果輸出到OBS
job_step_b = wf.steps.JobStep(
    name="training_job_b",  # 訓(xùn)練節(jié)點(diǎn)的名稱，命名規(guī)范(只能包含英文字母、數(shù)字、下劃線（_）、中劃線（-），并且只能以英文字母開頭，長度限制為64字符)，一個(gè)Workflow里的兩個(gè)step名稱不能重復(fù)
    title="圖像分類訓(xùn)練",  # 標(biāo)題信息，不填默認(rèn)使用name
    algorithm=wf.AIGalleryAlgorithm(
        subscription_id="subscription_id",  # 算法訂閱ID
        item_version_id="item_version_id",  # 算法訂閱版本ID，也可直接填寫版本號(hào)
        parameters=[]

    ),  # 訓(xùn)練使用的算法對(duì)象，示例中使用AIGallery訂閱的算法；部分算法超參的值如果無需修改，則在parameters字段中可以不填寫，系統(tǒng)自動(dòng)填充相關(guān)超參值
    inputs=wf.steps.JobInput(name="data_url", data=obs_data),
    outputs=[wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=storage.join("directory_path_b")))],
    spec=wf.steps.JobSpec(
        resource=wf.steps.JobResource(
            flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="訓(xùn)練資源規(guī)格")

        )
    ),  # 訓(xùn)練資源規(guī)格信息
    policy=wf.steps.StepPolicy(skip_conditions=condition_equal_b)
)

# 定義模型名稱參數(shù)
model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR)

model_step = wf.steps.ModelStep(
    name="model_registration",  # 模型注冊(cè)節(jié)點(diǎn)的名稱，命名規(guī)范(只能包含英文字母、數(shù)字、下劃線（_）、中劃線（-），并且只能以英文字母開頭，長度限制為64字符)，一個(gè)Workflow里的兩個(gè)step名稱不能重復(fù)
    title="模型注冊(cè)",  # 標(biāo)題信息
    inputs=wf.steps.ModelInput(name='model_input', data=wf.data.DataConsumptionSelector(data_list=[job_step_a.outputs["train_url"].as_input(), job_step_b.outputs["train_url"].as_input()])),  # 選擇job_step_a或者job_step_b的輸出作為輸入
    outputs=wf.steps.ModelOutput(name='model_output', model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow")),  # ModelStep的輸出
    depend_steps=[job_step_a, job_step_b],  # 依賴的作業(yè)類型節(jié)點(diǎn)對(duì)象
)

workflow = wf.Workflow(
    name="new-condition-demo",
    desc="this is a demo workflow",
    steps=[job_step_a, job_step_b, model_step],
    storages=storage
)