华为云用户手册

进阶示例 import modelarts.workflow as wf left_value = wf.Placeholder(name="left_value", placeholder_type=wf.PlaceholderType.BOOL, default=True) condition1 = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.EQ, left=left_value, right=True) internal_condition_1 = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.GT, left=10, right=9) internal_condition_2 = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.LT, left=10, right=9) # condition2的结果为internal_condition_1 || internal_condition_2 condition2 = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.OR, left=internal_condition_1, right=internal_condition_2) condition_step = wf.steps.ConditionStep( name="condition_step_test", # 条件节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 conditions=[condition1, condition2], # 条件对象,允许多个条件，条件之间的关系为&& if_then_steps=["job_step_1"], # 当condition结果为true时，名称为job_step_1的节点允许执行，名称为job_step_2的节点跳过不执行 else_then_steps=["job_step_2"] # 当condition结果为false时，名称为job_step_2的节点允许执行，名称为job_step_1的节点跳过不执行 ) # 该节点仅作为示例使用，其他字段需自行补充 job_step_1 = wf.steps.JobStep( name="job_step_1", depend_steps=condition_step ) # 该节点仅作为示例使用，其他字段需自行补充 job_step_2 = wf.steps.JobStep( name="job_step_2", depend_steps=condition_step ) workflow = wf.Workflow( name="condition-demo", desc="this is a demo workflow", steps=[condition_step, job_step_1, job_step_2], ) ConditionStep支持多条件节点的嵌套使用，用户可以基于不同的场景灵活设计。 条件节点只支持双分支的选择执行，局限性较大，推荐您使用新的分支功能，可以在不添加新节点的情况下完全覆盖ConditionStep的能力，详情请参见配置节点参数控制分支执行章节。

从训练作业中注册模型（模型输入来源JobStep的输出） import modelarts.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="训练资源规格") ) )# 训练资源规格信息 ) # 通过ModelStep来定义一个模型注册节点，输入来源于JobStep的输出 # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_registration = wf.steps.ModelStep( name="model_registration", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 inputs=wf.steps.ModelInput(name='model_input', data=job_step.outputs["train_url"].as_input()), # ModelStep的输入来源于依赖的JobStep的输出 outputs=wf.steps.ModelOutput(name='model_output',model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow")), # ModelStep的输出 depend_steps=job_step # 依赖的作业类型节点对象 ) # job_step是wf.steps.JobStep的 实例对象，train_url是wf.steps.JobOutput的name字段值 workflow = wf.Workflow( name="model-step-demo", desc="this is a demo workflow", steps=[job_step, model_registration], storages=[storage] )

从 自定义镜像 中注册模型 import modelarts.workflow as wf # 通过ModelStep来定义一个模型注册节点，输入来源于自定义镜像地址 # 定义镜像数据 swr = wf.data.SWRImagePlaceholder(name="placeholder_name") # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_registration = wf.steps.ModelStep( name="model_registration", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 inputs=wf.steps.ModelInput(name="input",data=swr), # ModelStep的输入在运行时配置；data字段的值也可使用wf.data.SWRImage(swr_path="fake_path")表示 outputs=wf.steps.ModelOutput(name='model_output',model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow"))# ModelStep的输出 ) workflow = wf.Workflow( name="model-step-demo", desc="this is a demo workflow", steps=[model_registration] )

从训练作业中注册模型（模型输入来源OBS路径，训练完成的模型已存储到OBS路径） import modelarts.workflow as wf # 通过ModelStep来定义一个模型注册节点，输入来源于OBS中 # 定义OBS数据对象 obs = wf.data.OBSPlaceholder(name = "obs_placeholder_name", object_type = "directory" ) # object_type必须是file或者directory # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_registration = wf.steps.ModelStep( name="model_registration", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 inputs=wf.steps.ModelInput(name='model_input', data=obs), # ModelStep的输入在运行时配置；data字段的值也可使用wf.data.OBSPath(obs_path="fake_obs_path")表示 outputs=wf.steps.ModelOutput(name='model_output',model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow"))# ModelStep的输出 ) workflow = wf.Workflow( name="model-step-demo", desc="this is a demo workflow", steps=[model_registration] )

使用模板的方式注册模型 import modelarts.workflow as wf # 通过ModelStep来定义一个模型注册节点，并通过预置模板进行注册 # 定义预置模板对象，Template对象中的字段可使用Placeholder表示 template = wf.steps.Template( template_id="fake_template_id", infer_format="fake_infer_format", template_inputs=[ wf.steps.TemplateInputs( input_id="fake_input_id", input="fake_input_file" ) ] ) # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_registration = wf.steps.ModelStep( name="model_registration", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 outputs=wf.steps.ModelOutput( name='model_output', model_config=wf.steps.ModelConfig( model_name=model_name, model_type="Template", template=template ) )# ModelStep的输出 ) workflow = wf.Workflow( name="model-step-demo", desc="this is a demo workflow", steps=[model_registration] )

使用自定义镜像+OBS的方式注册模型 import modelarts.workflow as wf # 通过ModelStep来定义一个模型注册节点，输入来源于自定义镜像地址 # 定义镜像数据 swr = wf.data.SWRImagePlaceholder(name="placeholder_name") # 定义OBS模型数据 model_obs = wf.data.OBSPlaceholder(name = "obs_placeholder_name", object_type = "directory" ) # object_type必须是file或者directory # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_registration = wf.steps.ModelStep( name="model_registration", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 inputs=[ wf.steps.ModelInput(name="input",data=swr), # ModelStep的输入在运行时配置；data字段的值也可使用wf.data.SWRImage(swr_path="fake_path")表示 wf.steps.ModelInput(name="input",data=model_obs) # ModelStep的输入在运行时配置；data字段的值也可使用wf.data.OBSPath(obs_path="fake_obs_path")表示 ], outputs=wf.steps.ModelOutput( name='model_output', model_config=wf.steps.ModelConfig( model_name=model_name, model_type="Custom", dynamic_load_mode="Single" ) )# ModelStep的输出 ) workflow = wf.Workflow( name="model-step-demo", desc="this is a demo orkflow", steps=[model_registration] )

使用案例 场景一：基于数据集发布版本 使用场景：当数据集更新了数据时，可以通过该节点发布新的数据集版本供后续的节点使用。 from modelarts import workflow as wf # 通过ReleaseDatasetStep将输入的数据集对象发布新的版本，输出带有版本信息的数据集对象 # 定义数据集对象 dataset = wf.data.DatasetPlaceholder(name="input_dataset") # 定义训练验证切分比参数 train_ration = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR, default="0.8") release_version = 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的输出 ) workflow = wf.Workflow( name="dataset-release-demo", desc="this is a demo workflow", steps=[release_version] ) 场景二：基于标注任务发布版本 当标注任务更新了数据或者标注信息时，可以通过该节点发布新的数据集版本供后续的节点使用。 from modelarts import workflow as wf # 通过ReleaseDatasetStep将输入的标注任务对象发布新的版本，输出带有版本信息的数据集对象 # 定义标注任务对象 label_task = wf.data.LabelTaskPlaceholder(name="label_task_placeholder_name") # 定义训练验证切分比参数 train_ration = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR, default="0.8") release_version = wf.steps.ReleaseDatasetStep( name="release_dataset", # 数据集发布节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集版本发布", # 标题信息，不填默认使用name值 inputs=wf.steps.ReleaseDatasetInput(name="input_name", data=label_task), # ReleaseDatasetStep的输入， 标注任务对象在运行时配置；data字段也可使用wf.data.LabelTask(dataset_name="dataset_name", task_name="label_task_name")表示 outputs=wf.steps.ReleaseDatasetOutput(name="output_name", dataset_version_config=wf.data.DatasetVersionConfig(train_evaluate_sample_ratio=train_ration)), # 数据集的训练验证切分比 ) workflow = wf.Workflow( name="dataset-release-demo", desc="this is a demo workflow", steps=[release_version] ) 场景三：基于数据集标注节点，构建数据集版本发布节点 使用场景：数据集标注节点的输出作为数据集版本发布节点的输入。 from modelarts import workflow as wf # 通过ReleaseDatasetStep将输入的标注任务对象发布新的版本，输出带有版本信息的数据集对象 # 定义训练验证切分比参数 train_ration = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR, default="0.8") release_version = wf.steps.ReleaseDatasetStep( name="release_dataset", # 数据集发布节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集版本发布", # 标题信息，不填默认使用name值 inputs=wf.steps.ReleaseDatasetInput(name="input_name", data=labeling_step.outputs["output_name"].as_input()), # ReleaseDatasetStep的输入， 标注任务对象在运行时配置；data字段也可使用wf.data.LabelTask(dataset_name="dataset_name", task_name="label_task_name")表示 outputs=wf.steps.ReleaseDatasetOutput(name="output_name", dataset_version_config=wf.data.DatasetVersionConfig(train_evaluate_sample_ratio=train_ration)), # 数据集的训练验证切分比 depend_steps = [labeling_step] # 依赖的数据集标注节点对象 ) # labeling_step是wf.steps.LabelingStep的实例对象，output_name是wf.steps.LabelingOutput的name字段值 workflow = wf.Workflow( name="dataset-release-demo", desc="this is a demo workflow", steps=[release_version] )

功能介绍 通过对ModelArts数据集能力进行封装，实现数据集的版本自动发布的功能。数据集版本发布节点主要用于将已存在的数据集或者标注任务进行版本发布，每个版本相当于数据的一个快照，可用于后续的数据溯源。主要应用场景如下： 对于数据标注这种操作，可以在标注完成后自动帮助用户发布新的数据集版本，结合as_input的能力提供给后续节点使用。 当模型训练需要更新数据时，可以使用数据集导入节点先导入新的数据，然后再通过该节点发布新的版本供后续节点使用。

属性总览 您可以使用ReleaseDatasetStep来构建数据集版本发布节点，ReleaseDatasetStep结构如下： 表1 ReleaseDatasetStep 属性 描述 是否必填 数据类型 name 数据集版本发布节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 是 str inputs 数据集版本发布节点的输入列表 是 ReleaseDatasetInput或者ReleaseDatasetInput的列表 outputs 数据集版本发布节点的输出列表 是 ReleaseDatasetOutput或者ReleaseDatasetOutput的列表 title title信息，主要用于前端的名称展示 否 str description 数据集版本发布节点的描述信息 否 str policy 节点执行的policy 否 StepPolicy depend_steps 依赖的节点列表 否 Step或者Step的列表 表2 ReleaseDatasetInput 属性 描述 是否必填 数据类型 name 数据集版本发布节点的输入名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输入名称不能重复 是 str data 数据集版本发布节点的输入数据对象 是 数据集或标注任务相关对象，当前仅支持Dataset，DatasetConsumption，DatasetPlaceholder，LabelTask，LabelTaskPlaceholder，LabelTaskConsumption，DataConsumptionSelector 表3 ReleaseDatasetOutput 属性 描述 是否必填 数据类型 name 数据集版本发布节点的输出名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输出名称不能重复 是 str dataset_version_config 数据集版本发布相关配置信息 是 DatasetVersionConfig 表4 DatasetVersionConfig 属性 描述 是否必填 数据类型 version_name 数据集版本名称，推荐使用类似V001的格式，不填则默认从V001往上递增。 否 str或者Placeholder version_format 版本格式，默认为"Default"，也可支持"CarbonData"。 否 str train_evaluate_sample_ratio 训练-验证集比例，默认值为"1.00"。取值范围为0-1.00， 例如"0.8"表示训练集比例为80%，验证集比例为20%。 否 str或者Placeholder clear_hard_property 是否清除难例，默认为“True”。 否 bool或者Placeholder remove_sample_usage 是否清除数据集已有的usage信息，默认为“True”。 否 bool或者Placeholder label_task_type 标注任务的类型。当输入是数据集时，该字段必填，用来指定数据集版本的标注场景。输入是标注任务时该字段不用填写。 否 LabelTaskTypeEnum 支持以下几种类型： IMAGE_CLASSIFICATION （图像分类） OBJECT_DETECTION = 1 （物体检测） IMAGE_SEGMENTATION （图像分割） TEXT_CLASSIFICATION （文本分类） NAMED_ENTITY_RECOGNITION （命名实体） TEXT_TRIPLE （文本三元组） AUDIO_CLASSIFICATION （声音分类） SPEECH_CONTENT （语音内容）SPEECH_SEGMENTATION （语音分割） TABLE （表格数据） VIDEO_ANNOTATION （视频标注） description 版本描述信息。 否 str 如果您没有特殊需求，则可直接使用内置的默认值，例如example = DatasetVersionConfig()

属性总览 您可以使用DatasetImportStep来构建数据集导入节点，DatasetImportStep结构如下。 表1 DatasetImportStep 属性 描述 是否必填 数据类型 name 数据集导入节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复。 是 str inputs 数据集导入节点的输入列表。 是 DatasetImportInput或者DatasetImportInput的列表 outputs 数据集导入节点的输出列表。 是 DatasetImportOutput或者DatasetImportOutput的列表 properties 数据集导入相关的配置信息。 是 ImportDataInfo title title信息，主要用于前端的名称展示。 否 str description 数据集导入节点的描述信息。 否 str policy 节点执行的policy。 否 StepPolicy depend_steps 依赖的节点列表。 否 Step或者Step的列表 表2 DatasetImportInput 属性 描述 是否必填 数据类型 name 数据集导入节点的输入名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输入名称不能重复。 是 str data 数据集导入节点的输入数据对象。 是 数据集、OBS或标注任务相关对象，当前仅支持Dataset，DatasetConsumption，DatasetPlaceholder，OBSPath，OBSConsumption，OBSPlaceholder，LabelTask，LabelTaskPlaceholder，LabelTaskConsumption，DataConsumptionSelector 表3 DatasetImportOutput 属性 描述 是否必填 数据类型 name 数据集导入节点的输出名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输出名称不能重复。 是 str 表4 ImportDataInfo 属性 描述 是否必填 数据类型 annotation_format_config 导入的标注格式的配置参数。 否 AnnotationFormatConfig excluded_labels 不导入包含指定标签的样本。 否 Label的列表 import_annotated 用于导入智能标注结果的任务，是否导入原数据集中已标注的样本到待确认，默认值为"false"即不导入原数据集中已标注的样本到待确认。可选值如下： true：导入原数据集中已标注的样本到待确认 false：不导入原数据集中已标注的样本到待确认 否 bool import_annotations 是否导入标签。可选值如下： true：导入标签（默认值） false：不导入标签 否 bool import_samples 是否导入样本。可选值如下： true：导入样本（默认值） false：不导入样本 否 bool import_type 导入方式。可选值如下： dir：目录导入 manifest：按manifest文件导入 否 ImportTypeEnum included_labels 导入包含指定标签的样本。 否 Label的列表 label_format 标签格式，此参数仅文本类数据集使用。 否 LabelFormat 表5 AnnotationFormatConfig 属性 描述 是否必填 数据类型 format_name 标注格式的名称。 否 AnnotationFormatEnum parameters 标注格式的高级参数。 否 AnnotationFormatParameters scene 标注场景，可选参数。 否 LabelTaskTypeEnum 表6 AnnotationFormatParameters 属性 描述 是否必填 数据类型 difficult_only 是否只导入难例。可选值如下： true：只导入难例样本 false：导入全部样本（默认值） 否 bool included_labels 导入包含指定标签的样本。 否 Label的列表 label_separator 标签与标签之间的分隔符，默认为逗号分隔，分隔符需转义。分隔符仅支持一个字符，必须为大小写字母，数字和“!@#$%^&*_=|?/':.;,”其中的某一字符。 否 str sample_label_separator 文本与标签之间的分隔符，默认为Tab键分隔，分隔符需转义。分隔符仅支持一个字符，必须为大小写字母，数字和“!@#$%^&*_=|?/':.;,”其中的某一字符。 否 str

使用案例 主要包含三种场景的用例： 场景一：将指定存储路径下的数据导入到目标数据集中。适用于需要对数据集进行数据更新的操作。 用户将指定路径下已标注的数据导入到数据集中（同时导入标签信息），后续可增加数据集版本发布节点进行版本发布。 数据准备：提前在ModelArts管理控制台，创建数据集，并将已标注的数据上传至OBS中。 from modelarts import workflow as wf # 通过DatasetImportStep将指定路径下的数据导入到数据集中，输出数据集对象 # 定义数据集对象 dataset = wf.data.DatasetPlaceholder(name="input_dataset") # 定义OBS数据对象 obs = wf.data.OBSPlaceholder(name = "obs_placeholder_name", object_type = "directory" ) # object_type必须是file或者directory dataset_import = wf.steps.DatasetImportStep( name="data_import", # 数据集导入节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集导入", # 标题信息，不填默认使用name值 inputs=[ wf.steps.DatasetImportInput(name="input_name_1", data=dataset), # 目标数据集在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="dataset_name")表示 wf.steps.DatasetImportInput(name="input_name_2", data=obs) # 导入的数据存储路径，运行时配置；data字段也可使用wf.data.OBSPath(obs_path="obs_path")表示 ],# DatasetImportStep的输入 outputs=wf.steps.DatasetImportOutput(name="output_name"), # DatasetImportStep的输出 properties=wf.steps.ImportDataInfo( annotation_format_config=[ wf.steps.AnnotationFormatConfig( format_name=wf.steps.AnnotationFormatEnum.MA_IMAGE_CLASSIFICATION_V1, # 已标注数据的标注格式，示例为图像分类 scene=wf.data.LabelTaskTypeEnum.IMAGE_CLASSIFICATION # 标注的场景类型 ) ] ) ) workflow = wf.Workflow( name="dataset-import-demo", desc="this is a demo workflow", steps=[dataset_import] ) 用户将指定路径下未标注的数据导入到数据集中，后续可增加数据集标注节点对新增数据进行标注。 数据准备：提前在ModelArts界面创建数据集，并将未标注的数据上传至OBS中。 from modelarts import workflow as wf # 通过DatasetImportStep将指定路径下的数据导入到数据集中，输出数据集对象 # 定义数据集对象 dataset = wf.data.DatasetPlaceholder(name="input_dataset") # 定义OBS数据对象 obs = wf.data.OBSPlaceholder(name = "obs_placeholder_name", object_type = "directory" ) # object_type必须是file或者directory dataset_import = wf.steps.DatasetImportStep( name="data_import", # 数据集导入节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集导入", # 标题信息，不填默认使用name值 inputs=[ wf.steps.DatasetImportInput(name="input_name_1", data=dataset), # 目标数据集在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="dataset_name")表示 wf.steps.DatasetImportInput(name="input_name_2", data=obs) # 导入的数据存储路径，运行时配置；data字段也可使用wf.data.OBSPath(obs_path="obs_path")表示 ],# DatasetImportStep的输入 outputs=wf.steps.DatasetImportOutput(name="output_name") # DatasetImportStep的输出 ) workflow = wf.Workflow( name="dataset-import-demo", desc="this is a demo workflow", steps=[dataset_import] ) 场景二：将指定存储路径下的数据导入到指定标注任务中。适用于需要对标注任务进行数据更新的操作。 用户将指定路径下已标注的数据导入到标注任务中（同时导入标签信息），后续可增加数据集版本发布节点进行版本发布。 数据准备：基于使用的数据集，提前创建标注任务，并将已标注的数据上传至OBS中。 from modelarts import workflow as wf # 通过DatasetImportStep将指定路径下的数据导入到标注任务中，输出标注任务对象 # 定义标注任务对象 label_task = wf.data.LabelTaskPlaceholder(name="label_task_placeholder_name") # 定义OBS数据对象 obs = wf.data.OBSPlaceholder(name = "obs_placeholder_name", object_type = "directory" ) # object_type必须是file或者directory dataset_import = wf.steps.DatasetImportStep( name="data_import", # 数据集导入节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集导入", # 标题信息，不填默认使用name值 inputs=[ wf.steps.DatasetImportInput(name="input_name_1", data=label_task), # 目标标注任务对象，在运行时配置；data字段也可使用wf.data.LabelTask(dataset_name="dataset_name", task_name="label_task_name")表示 wf.steps.DatasetImportInput(name="input_name_2", data=obs) # 导入的数据存储路径，运行时配置；data字段也可使用wf.data.OBSPath(obs_path="obs_path")表示 ],# DatasetImportStep的输入 outputs=wf.steps.DatasetImportOutput(name="output_name"), # DatasetImportStep的输出 properties=wf.steps.ImportDataInfo( annotation_format_config=[ wf.steps.AnnotationFormatConfig( format_name=wf.steps.AnnotationFormatEnum.MA_IMAGE_CLASSIFICATION_V1, # 已标注数据的标注格式，示例为图像分类 scene=wf.data.LabelTaskTypeEnum.IMAGE_CLASSIFICATION # 标注的场景类型 ) ] ) ) workflow = wf.Workflow( name="dataset-import-demo", desc="this is a demo workflow", steps=[dataset_import] ) 用户将指定路径下未标注的数据导入到标注任务中，后续可增加数据集标注节点对新增数据进行标注。 数据准备：基于使用的数据集，提前创建标注任务，并将未标注的数据上传至OBS中。 from modelarts import workflow as wf # 通过DatasetImportStep将指定路径下的数据导入到标注任务中，输出标注任务对象 # 定义标注任务对象 label_task = wf.data.LabelTaskPlaceholder(name="label_task_placeholder_name") # 定义OBS数据对象 obs = wf.data.OBSPlaceholder(name = "obs_placeholder_name", object_type = "directory" ) # object_type必须是file或者directory dataset_import = wf.steps.DatasetImportStep( name="data_import", # 数据集导入节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集导入", # 标题信息，不填默认使用name值 inputs=[ wf.steps.DatasetImportInput(name="input_name_1", data=label_task), # 目标标注任务对象，在运行时配置；data字段也可使用wf.data.LabelTask(dataset_name="dataset_name", task_name="label_task_name")表示 wf.steps.DatasetImportInput(name="input_name_2", data=obs) # 导入的数据存储路径，运行时配置；data字段也可使用wf.data.OBSPath(obs_path="obs_path")表示 ],# DatasetImportStep的输入 outputs=wf.steps.DatasetImportOutput(name="output_name") # DatasetImportStep的输出 ) workflow = wf.Workflow( name="dataset-import-demo", desc="this is a demo workflow", steps=[dataset_import] ) 场景三：基于数据集创建节点构建数据集导入节点。数据集创建节点的输出作为数据集导入节点的输入。 from modelarts import workflow as wf # 通过DatasetImportStep将指定路径下的数据导入到数据集中，输出数据集对象 # 定义OBS数据对象 obs = wf.data.OBSPlaceholder(name = "obs_placeholder_name", object_type = "directory" ) # object_type必须是file或者directory dataset_import = wf.steps.DatasetImportStep( name="data_import", # 数据集导入节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集导入", # 标题信息，不填默认使用name值 inputs=[ wf.steps.DatasetImportInput(name="input_name_1", data=create_dataset.outputs["create_dataset_output"].as_input()), # 数据集创建节点的输出作为导入节点的输入 wf.steps.DatasetImportInput(name="input_name_2", data=obs) # 导入的数据存储路径，运行时配置；data字段也可使用wf.data.OBSPath(obs_path="obs_path")表示 ],# DatasetImportStep的输入 outputs=wf.steps.DatasetImportOutput(name="output_name"), # DatasetImportStep的输出 depend_steps=create_dataset # 依赖的数据集创建节点对象 ) # create_dataset是 wf.steps.CreateDatasetStep的一个实例，create_dataset_output是wf.steps.CreateDatasetOutput的name字段值 workflow = wf.Workflow( name="dataset-import-demo", desc="this is a demo workflow", steps=[dataset_import] )

功能介绍 通过对ModelArts数据集能力进行封装，实现数据集的数据导入功能。数据集导入节点主要用于将指定路径下的数据导入到数据集或者标注任务中，主要应用场景如下： 适用于数据不断迭代的场景，可以将一些新增的原始数据或者已标注数据导入到标注任务中，并通过后续的数据集标注节点进行标注。 对于一些已标注好的原始数据，可以直接导入到数据集或者标注任务中，并通过后续的数据集版本发布节点获取带有版本信息的数据集对象。

Workflow数据集标注节点代码样例 主要包含三种场景的用例： 场景一：基于用户指定的数据集创建标注任务，并等待用户标注完成。 使用场景： 用户只创建了一个未标注完成的数据集，需要在工作流运行时对数据进行人工标注。 可以放在数据集导入节点之后，对导入的新数据进行人工标注。 数据准备：提前在ModelArts管理控制台创建一个数据集。 from modelarts import workflow as wf # 通过LabelingStep给输入的数据集对象创建新的标注任务，并等待用户标注完成 # 定义输入的数据集对象 dataset = wf.data.DatasetPlaceholder(name="input_dataset") # 定义标注任务的名称参数 task_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) labeling = wf.steps.LabelingStep( name="labeling", # 数据集标注节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集标注", # 标题信息，不填默认使用name值 properties=wf.steps.LabelTaskProperties( task_type=wf.data.LabelTaskTypeEnum.IMAGE_CLASSIFICATION, # 标注任务的类型，以图像分类为例 task_name=task_name # 该名称对应的标注任务如果不存在则创建，如果存在则直接使用该任务 ), inputs=wf.steps.LabelingInput(name="input_name", data=dataset), # LabelingStep的输入，数据集对象在运行时配置；data字段也可使用wf.data.Dataset(dataset_name="fake_dataset_name")表示 outputs=wf.steps.LabelingOutput(name="output_name"), # LabelingStep的输出 ) workflow = wf.Workflow( name="labeling-step-demo", desc="this is a demo workflow", steps=[labeling] ) 场景二：基于指定的标注任务进行标注。 使用场景： 用户基于数据集自主创建了一个标注任务，需要在工作流运行时对数据进行人工标注。 可以放在数据集导入节点之后，对导入的新数据进行人工标注。 数据准备：提前在ModelArts管理控制台，基于使用的数据集创建一个标注任务。 from modelarts import workflow as wf # 用户输入标注任务，等待用户标注完成 # 定义数据集的标注任务对象 label_task = wf.data.LabelTaskPlaceholder(name="label_task_placeholder_name") labeling = wf.steps.LabelingStep( name="labeling", # 数据集标注节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集标注", # 标题信息，不填默认使用name值 inputs=wf.steps.LabelingInput(name="input_name", data=label_task), # LabelingStep的输入，标注任务对象在运行时配置；data字段也可使用wf.data.LabelTask(dataset_name="dataset_name", task_name="label_task_name")来表示 outputs=wf.steps.LabelingOutput(name="output_name"), # LabelingStep的输出 ) workflow = wf.Workflow( name="labeling-step-demo", desc="this is a demo workflow", steps=[labeling] ) 场景三：基于数据集创建节点的输出创建标注任务。 使用场景：数据集创建节点的输出作为数据集数据标注节点的输入。 from modelarts import workflow as wf # 定义数据集输出路径参数 dataset_output_path = wf.Placeholder(name="dataset_output_path", placeholder_type=wf.PlaceholderType.STR, placeholder_format="obs") # 定义数据集名称参数 dataset_name = wf.Placeholder(name="dataset_name", placeholder_type=wf.PlaceholderType.STR) create_dataset = wf.steps.CreateDatasetStep( name="create_dataset",# 数据集创建节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集创建", # 标题信息，不填默认使用name值 inputs=wf.steps.CreateDatasetInput(name="input_name", data=wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory")),# CreateDatasetStep的输入，数据在运行时进行配置；data字段也可使用wf.data.OBSPath(obs_path="fake_obs_path")对象表示 outputs=wf.steps.CreateDatasetOutput(name="create_dataset_output", config=wf.data.OBSOutputConfig(obs_path=dataset_output_path)),# CreateDatasetStep的输出 properties=wf.steps.DatasetProperties( dataset_name=dataset_name, # 该名称对应的数据集如果不存在，则创建新的数据集;如果已存在，则直接使用该名称对应的数据集 data_type=wf.data.DataTypeEnum.IMAGE, # 数据集对应的数据类型, 示例为图像 ) ) # 定义标注任务的名称参数 task_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) labeling = wf.steps.LabelingStep( name="labeling", # 数据集标注节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="数据集标注", # 标题信息，不填默认使用name值 properties=wf.steps.LabelTaskProperties( task_type=wf.data.LabelTaskTypeEnum.IMAGE_CLASSIFICATION, # 标注任务的类型，以图像分类为例 task_name=task_name # 该名称对应的标注任务如果不存在则创建，如果存在则直接使用该任务 ), inputs=wf.steps.LabelingInput(name="input_name", data=create_dataset.outputs["create_dataset_output"].as_input()), # LabelingStep的输入，data数据来源为数据集创建节点的输出 outputs=wf.steps.LabelingOutput(name="output_name"), # LabelingStep的输出 depend_steps=create_dataset # 依赖的数据集创建节点对象 ) # create_dataset是 wf.steps.CreateDatasetStep的一个实例，create_dataset_output是wf.steps.CreateDatasetOutput的name字段值 workflow = wf.Workflow( name="labeling-step-demo", desc="this is a demo workflow", steps=[create_dataset, labeling] )

使用案例 统一存储主要用于JobStep中，下面代码示例全部以单训练节点为例。 from modelarts import workflow as wf # 构建一个InputStorage对象, 并且假设配置的根目录为"/root/input-data/" input_storage = wf.data.InputStorage(name="input_storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 # 构建一个OutputStorage对象, 并且假设配置的根目录为"/root/output/" output_storage = wf.data.OutputStorage(name="output_storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 # 通过JobStep来定义一个训练节点，输入数据来源为OBS，并将训练结果输出到OBS中 job_step = wf.steps.JobStep( name="training_job", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="图像分类训练", # 标题信息，不填默认使用name algorithm=wf.AIGalleryAlgorithm(subscription_id="subscription_ID", item_version_id="item_version_ID"), # 训练使用的算法对象，示例中使用AIGallery订阅的算法 inputs=[ wf.steps.JobInput(name="data_url_1", data=wf.data.OBSPath(obs_path = input_storage.join("/dataset1/new.manifest"))), # 获得的路径为：/root/input-data/dataset1/new.manifest wf.steps.JobInput(name="data_url_2", data=wf.data.OBSPath(obs_path = input_storage.join("/dataset2/new.manifest"))) # 获得的路径为：/root/input-data/dataset2/new.manifest ], outputs=wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=output_storage.join("/model/"))), # 训练输出的路径为：/root/output/执行ID/model/ spec=wf.steps.JobSpec( resource=wf.steps.JobResource( flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格") ), log_export_path=wf.steps.job_step.LogExportPath(obs_url=output_storage.join("/logs/")) # 日志输出的路径为：/root/output/执行ID/logs/ )# 训练资源规格信息 ) # 定义一个只包含job_step的工作流 workflow = wf.Workflow( name="test-workflow", desc="this is a test workflow", steps=[job_step], storages=[input_storage, output_storage] # 注意在整个工作流中使用到的Storage对象需要在这里添加 )

功能介绍 统一存储主要用于工作流的目录管理，帮助用户统一管理一个工作流中的所有存储路径，主要分为以下两个功能： 输入目录管理：开发者在编辑开发工作流时可以对所有数据的存储路径做统一管理，规定用户按照自己的目录规划来存放数据，而存储的根目录可以根据用户自己的需求自行配置。该方式只做目录的编排，不会自动创建新的目录。 输出目录管理：开发者在编辑开发工作流时可以对所有的输出路径做统一管理，用户无需手动创建输出目录，只需要在工作流运行前配置存储根路径，并且可以根据开发者的目录编排规则在指定目录下查看输出的数据信息。此外同一个工作流的多次运行支持输出到不同的目录下，对不同的执行做了很好的数据隔离。

常用方式 InputStorage（路径拼接） 该对象主要用于帮助用户统一管理输入的目录，使用示例如下： import modelarts.workflow as wf storage = wf.data.InputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 input_data = wf.data.OBSPath(obs_path = storage.join("directory_path")) # 注意，如果是目录则最后需要加"/"，例如：storage.join("/input/data/") 工作流运行时，如果storage对象配置的根路径为"/root/"，则最后得到的路径为"/root/directory_path" OutputStorage（目录创建） 该对象主要用于帮助用户统一管理输出的目录，保证工作流每次执行输出到新目录，使用示例如下： import modelarts.workflow as wf storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 output_path = wf.data.OBSOutputConfig(obs_path = storage.join("directory_path")) # 注意，只能创建目录，不能创建文件 。 工作流运行时，如果storage对象配置的根路径为"/root/"，则系统自动创建相对目录，最后得到的路径为"/root/执行ID/directory_path"

进阶用法 Storage 该对象是InputStorage和OutputStorage的基类，包含了两者的所有能力，可以供用户灵活使用。 属性 描述 是否必填 数据类型 name 名称。 是 str title 不填默认使用name的值。 否 str description 描述信息。 否 str create_dir 表示是否自动创建目录，默认为“False”。 否 bool with_execution_id 表示创建目录时是否拼接execution_id，默认为“False”。该字段只有在create_dir为True时才支持设置为True。 否 bool 使用示例如下： 实现InputStorage相同的能力 import modelarts.workflow as wf # 构建一个Storage对象, with_execution_id=False, create_dir=False storage = wf.data.Storage(name="storage_name", title="title_info", description="description_info", with_execution_id=False, create_dir=False) input_data = wf.data.OBSPath(obs_path = storage.join("directory_path")) # 注意，如果是目录则最后需要加"/"，例如：storage.join("/input/data/") 工作流运行时，如果storage对象配置的根路径为"/root/"，则最后得到的路径为"/root/directory_path" 实现OutputStorage相同的能力 import modelarts.workflow as wf # 构建一个Storage对象, with_execution_id=True, create_dir=True storage = wf.data.Storage(name="storage_name", title="title_info", description="description_info", with_execution_id=True, create_dir=True) output_path = wf.data.OBSOutputConfig(obs_path = storage.join("directory_path")) # 注意，只能创建目录，不能创建文件 工作流运行时，如果storage对象配置的根路径为"/root/"，则系统自动创建相对目录，最后得到的路径为"/root/执行ID/directory_path" 通过join方法的参数实现同一个Storage的不同用法 import modelarts.workflow as wf # 构建一个Storage对象, 并且假设Storage配置的根目录为"/root/" storage = wf.data.Storage(name="storage_name", title="title_info", description="description_info", with_execution_id=False, create_dir=False) input_data1 = wf.data.OBSPath(obs_path = storage) # 得到的路径为：/root/ input_data2 = wf.data.OBSPath(obs_path = storage.join("directory_path")) # 得到的路径为：/root/directory_path，需要用户自行保证该路径存在 output_path1 = wf.data.OBSOutputConfig(obs_path = storage.join(directory="directory_path", with_execution_id=False, create_dir=True)) # 系统自动创建目录，得到的路径为：/root/directory_path output_path2 = wf.data.OBSOutputConfig(obs_path = storage.join(directory="directory_path", with_execution_id=True, create_dir=True)) # 系统自动创建目录，得到的路径为：/root/执行ID/directory_path Storage可实现链式调用 使用示例如下： import modelarts.workflow as wf # 构建一个基类Storage对象, 并且假设Storage配置的根目录为"/root/" storage = wf.data.Storage(name="storage_name", title="title_info", description="description_info", with_execution_id=False, create_dir=Fals) input_storage = storage.join("directory_path_1") # 得到的路径为：/root/directory_path_1 input_storage_next = input_storage.join("directory_path_2") # 得到的路径为： /root/directory_path_1/directory_path_2

使用案例 int类型参数 from modelarts import workflow as wf wf.Placeholder(name="placeholder_int", placeholder_type=wf.PlaceholderType.INT, default=1, description="这是一个int类型的参数") str类型参数 from modelarts import workflow as wf wf.Placeholder(name="placeholder_str", placeholder_type=wf.PlaceholderType.STR, default="default_value", description="这是一个str类型的参数") bool类型参数 from modelarts import workflow as wf wf.Placeholder(name="placeholder_bool", placeholder_type=wf.PlaceholderType.BOOL, default=True, description="这是一个bool类型的参数") float类型参数 from modelarts import workflow as wf wf.Placeholder(name="placeholder_float", placeholder_type=wf.PlaceholderType.FLOAT, default=0.1, description="这是一个float类型的参数") Enum类型参数 from modelarts import workflow as wf wf.Placeholder(name="placeholder_enum", placeholder_type=wf.PlaceholderType.ENUM, default="a", enum_list=["a", "b"], description="这是一个enum类型的参数") dict类型参数 from modelarts import workflow as wf wf.Placeholder(name="placeholder_dict", placeholder_type=wf.PlaceholderType.JSON, default={"key": "value"}, description="这是一个dict类型的参数") list类型参数 from modelarts import workflow as wf wf.Placeholder(name="placeholder_list", placeholder_type=wf.PlaceholderType.LIST, default=[1, 2], placeholder_format="int", description="这是一个list类型的参数，并且value类型为int")

步骤四：发布Workflow至运行态 发布Workflow至运行态完成后需要进行配置，详细参考2。更多发布方式请参考发布Workflow。 Workflow开发完成后，执行如下代码发布Workflow至运行态。日志打印显示发布成功，则可前往ModelArts的Workflow页面中查看新发布的工作流。 workflow.release() 图3 发布成功 前往ModelArts控制台去查看新发布的Workflow，进入Workflow详情，单击“配置”进行参数配置。工作流相关的配置执行操作可参考如何使用Workflow。 图4 发布工作流

Data 数据对象用于节点的输入，主要可分为以下三种类型： 真实的数据对象，在工作流构建时直接指定： Dataset：用于定义已有的数据集，常用于数据标注，模型训练等场景 LabelTask: 用于定义已有的标注任务，常用于数据标注，数据集版本发布等场景 OBSPath：用于定义指定的OBS路径，常用于模型训练，数据集导入，模型导入等场景 ServiceData：用于定义一个已有的服务，只用于服务更新的场景 SWRImage：用于定义已有的SWR路径，常用于模型注册场景 GalleryModel：用于定义从gallery订阅的模型，常用于模型注册场景 占位符式的数据对象，在工作流运行时指定： DatasetPlaceholder：用于定义在运行时需要确定的数据集，对应Dataset对象，常用于数据标注，模型训练等场景 LabelTaskPlaceholder：用于定义在运行时需要确定的标注任务，对应LabelTask对象，常用于数据标注，数据集版本发布等场景 OBSPlaceholder：用于定义在运行时需要确定的OBS路径，对应OBSPath对象，常用于模型训练，数据集导入，模型导入等场景 ServiceUpdatePlaceholder：用于定义在运行时需要确定的已有服务，对应ServiceData对象，只用于服务更新的场景 SWRImagePlaceholder：用于定义在运行时需要确定的SWR路径，对应SWRImage对象，常用于模型注册场景 ServiceInputPlaceholder：用于定义在运行时需要确定服务部署所需的模型相关信息，只用于服务部署及服务更新场景 DataSelector：支持多种数据类型的选择，当前仅支持在JobStep节点中使用（仅支持选择OBS或者数据集） 数据选择对象： DataConsumptionSelector：用于在多个依赖节点的输出中选择一个有效输出作为数据输入，常用于存在条件分支的场景中（在构建工作流时未能确定数据输入来源为哪个依赖节点的输出，需根据依赖节点的实际执行情况进行自动选择） 表4 Dataset 属性 描述 是否必填 数据类型 dataset_name 数据集名称 是 str version_name 数据集版本名称 否 str 示例： example = Dataset(dataset_name = "**", version_name = "**") # 通过ModelArts的数据集，获取对应的数据集名称及相应的版本名称。 当Dataset对象作为节点的输入时，需根据业务需要自行决定是否填写version_name字段（比如LabelingStep、ReleaseDatasetStep不需要填写，JobStep必须填写）。 表5 LabelTask 属性 描述 是否必填 数据类型 dataset_name 数据集名称 是 str task_name 标注任务名称 是 str 示例： example = LabelTask(dataset_name = "**", task_name = "**") # 通过ModelArts的新版数据集，获取对应的数据集名称及相应的标注任务名称 表6 OBSPath 属性 描述 是否必填 数据类型 obs_path OBS路径 是 str，Storage 示例： example = OBSPath(obs_path = "**") # 通过 对象存储服务 ，获取已存在的OBS路径值 表7 ServiceData 属性 描述 是否必填 数据类型 service_id 服务的ID 是 str 示例： example = ServiceData(service_id = "**") # 通过ModelArts的在线服务，获取对应服务的服务ID，描述指定的在线服务。用于服务更新的场景。 表8 SWRImage 属性 描述 是否必填 数据类型 swr_path 容器镜像的SWR路径 是 str 示例： example = SWRImage(swr_path = "**") # 容器镜像地址，用于模型注册节点的输入 表9 GalleryModel 属性 描述 是否必填 数据类型 subscription_id 订阅模型的订阅ID 是 str version_num 订阅模型的版本号 是 str 示例： example = GalleryModel(subscription_id="**", version_num="**") # 订阅的模型对象，用于模型注册节点的输入 表10 DatasetPlaceholder 属性 描述 是否必填 数据类型 name 名称 是 str data_type 数据类型 否 DataTypeEnum delay 标志数据对象是否在节点运行时配置，默认为False 否 bool default 数据对象的默认值 否 Dataset 示例： example = DatasetPlaceholder(name = "**", data_type = DataTypeEnum.IMAGE_CLASSIFICATION) # 数据集对象的占位符形式，可以通过指定data_type限制数据集的数据类型 表11 OBSPlaceholder 属性 描述 是否必填 数据类型 name 名称 是 str object_type 表示OBS对象类型，仅支持"file"或者"directory" 是 str delay 标志数据对象是否在节点运行时配置，默认为False 否 bool default 数据对象的默认值 否 OBSPath 示例： example = OBSPlaceholder(name = "**", object_type = "directory" ) # OBS对象的占位符形式，object_type只支持两种类型， "file" 以及 "directory" 表12 LabelTaskPlaceholder 属性 描述 是否必填 数据类型 name 名称 是 str task_type 表示标注任务的类型 否 LabelTaskTypeEnum delay 标志数据对象是否在节点运行时配置，默认为False 否 bool 示例： example = LabelTaskPlaceholder(name = "**") # LabelTask对象的占位符形式 表13 ServiceUpdatePlaceholder 属性 描述 是否必填 数据类型 name 名称 是 str delay 标志数据对象是否在节点运行时配置，默认为False 否 bool 示例： example = ServiceUpdatePlaceholder(name = "**") # ServiceData对象的占位符形式，用于服务更新节点的输入 表14 SWRImagePlaceholder 属性 描述 是否必填 数据类型 name 名称 是 str delay 标志数据对象是否在节点运行时配置，默认为False 否 bool 示例： example = SWRImagePlaceholder(name = "**" ) # SWRImage对象的占位符形式，用于模型注册节点的输入 表15 ServiceInputPlaceholder 属性 描述 是否必填 数据类型 name 名称 是 str model_name 模型名称 是 str或者Placeholder model_version 模型版本 否 str envs 环境变量 否 dict delay 服务部署相关信息是否在节点运行时配置，默认为True 否 bool 示例： example = ServiceInputPlaceholder(name = "**" , model_name = "model_name") # 用于服务部署或者服务更新节点的输入 表16 DataSelector 属性 描述 是否必填 数据类型 name 名称 是 str data_type_list 支持的数据类型列表，当前仅支持obs、dataset 是 list delay 标志数据对象是否在节点运行时配置，默认为False 否 bool 示例： example = DataSelector(name = "**" ,data_type_list=["obs", "dataset"]) # 用于作业类型节点的输入 表17 DataConsumptionSelector 属性 描述 是否必填 数据类型 data_list 依赖节点的输出数据对象列表 是 list 示例： example = DataConsumptionSelector(data_list=[step1.outputs["step1_output_name"].as_input(), step2.outputs["step2_output_name"].as_input()]) # 从step1以及step2中选择有效输出作为输入，当step1跳过无输出，step2执行有输出时，将step2的有效输出作为输入（需保证data_list中同时只有一个有效输出）

Workflow Workflow是一个有向无环图（Directed Acyclic Graph，DAG），由节点和节点之间的关系描述组成。 图1 Workflow介绍 节点与节点之间的依赖关系由单箭头的线段来表示，依赖关系决定了节点的执行顺序，示例中的工作流在启动后将从左往右顺序执行。DAG也支持多分支结构，用户可根据实际场景进行灵活设计，在多分支场景下，并行分支的节点支持并行运行，具体请参考配置多分支节点数据章节。 表1 Workflow 属性 描述 是否必填 数据类型 name 工作流的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64位字符 是 str desc 工作流的描述信息 是 str steps 工作流包含的节点列表 是 list[Step] storages 统一存储对象列表 否 Storage或者list[Storage] policy 工作流的配置策略，主要用于部分运行场景 否 Policy

Step Step是组成Workflow的最小单元，体现在DAG中就是一个一个的节点，不同的Step类型承载了不同的服务能力，主要构成如下。 表2 Step 属性 描述 是否必填 数据类型 name 节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符 是 str title 节点的标题信息，主要用于在DAG中的展示，如果该字段未填写，则默认使用name进行展示 否 str step_type 节点的类型，决定了节点的功能 是 enum inputs 节点的输入列表 否 AbstractInput或者list[AbstractInput] outputs 节点的输出列表 否 AbstractOutput或者list[AbstractOutput] properties 节点的属性信息 否 dict policy 节点的执行策略，主要包含节点调度运行的时间间隔、节点执行的超时时间、以及节点执行是否跳过的相关配置 否 StepPolicy depend_steps 依赖节点的列表，该字段决定了DAG的结构，也决定了节点执行的顺序 否 Step或者list[Step] 表3 StepPolicy 属性 描述 是否必填 数据类型 poll_interval_seconds 节点调度时间周期，默认为1秒 是 str max_execution_minutes 节点运行超时时间，默认为10080分钟，即7天 是 str skip_conditions 节点是否跳过的条件列表 否 Condition或者Condition列表 Step是节点的超类，主要用于概念上的承载，用户不直接使用。根据功能的不同，构建了不同类型的节点，主要包括CreateDatasetStep、LabelingStep、DatasetImportStep、ReleaseDatasetStep、JobStep、ModelStep、ServiceStep、ConditionStep等，详情请见创建Workflow节点。

重试/停止/继续运行Workflow节点 重试 当单个节点运行失败时，用户可以通过重试按钮重新执行当前节点，无需重新启动工作流。在当前节点的运行状况页面，单击“重试”。在重试之前您也可以前往权限管理页面修改配置，节点重试启动后新修改的配置信息可以在当前执行中立即生效。 停止 单击指定节点查看详情，可以对运行中的节点进行停止操作。 继续运行 对于单个节点中设置了需要运行中配置的参数时，节点运行会处于“等待操作”状态，用户完成相关数据的配置后，可单击“继续运行”按钮并确认继续执行当前节点。

部分运行Workflow节点 针对大型、复杂的Workflow，为节省重复运行消耗的时间，在运行业务场景时，用户可以选择其中的部分节点作为业务场景运行，工作流在执行时将会按顺序执行部分运行节点。 部分运行Workflow节点，首先在新开发Workflow时，需要预先定义好部分运行场景。具体流程如下： 通过SDK创建工作流时，预先定义好部分运行场景，具体可参考在Workflow中指定仅运行部分节点。 在配置工作流时，打开“部分运行”开关，选择需要执行的部分运行场景，并填写完善相关节点的参数。 图1 部分运行 保存上一步的配置后，单击“启动”按钮即可启动部分运行场景。

查看Workflow工作流运行记录 运行记录是展示某条工作流所有运行状态数据的地方。 在Workflow列表页，单击某条工作流的名称，进入该工作流的详情页面。 在工作流的详情页，左侧区域即为该条工作流的所有运行记录。 图1 查看运行记录 您可以对当前工作流的所有运行记录，进行删除、编辑以及重新运行的操作。 删除：如果该条运行记录不再需要，您可以单击“删除”，在弹出的确认框中单击“确定”即可完成运行记录的删除。 编辑：如果您想对您当前的工作流下的所有运行记录进行区分，您可以单击“编辑”，对每一条运行记录添加相应的标签予以区分。 重新运行：可以单击“重新运行”直接在某条记录上运行该工作流。 您可以对该条工作流的所有运行记录进行筛选和对比。 筛选：该功能支持您对所有运行记录按照“运行状态”和“运行标签”进行筛选。 图2 筛选 对比：针对某条工作流的所有运行记录，按照状态、运行记录、启动时间、运行时长、参数等进行对比。 图3 对比 当单击“启动”运行工作流时，运行记录列表会自动刷新，并更新至最新一条的执行记录数据，且与DAG图和总览数据面板双向联动更新数据。每次启动后都会新增一条运行记录。 用户可以单击Workflow详情页中任一节点查询节点运行状况。包括节点的属性（节点的运行状态、启动时间以及运行时长）、输入位置与输出位置以及参数（数据集的标注任务名称）。 图4 查看节点运行情况 父主题： 管理Workflow

特性使用操作 安装优雅退出二进制包 通过ma_pre_start.sh安装whl包。 echo "[ma-pre-start] Enter the input directory" cd/home/ma-user/modelarts/inputs/data_url_0/ echo "[ma-pre-start] Start to install mindx-elastic 0.0.1版本" export PATH=/home/ma-user/anaconda/bin:$PATH pip install ./mindx_elastic-0.0.1-py3-none-any.whl echo "[ma-pre-start] Clean run package" sudo rm -rf ./script ./*.run ./run_package *.whl echo "[ma-pre-start] Set ENV" export G LOG _v=1 # 当前使用诊断模式需要用户手动设置成INFO日志级别 echo "[ma-pre-start] End" 创建训练任务 约束：MindSpore版本要求1.6.0及以上。 修改样例代码，增加如下内容： # 载入依赖接口 from mindx_elastic.terminating_message import ExceptionCheckpoint ... if args_opt.do_train: dataset = create_dataset() loss_cb = LossMonitor() cb = [loss_cb] if int(os.getenv('RANK_ID')) == 0: batch_num = dataset.get_dataset_size() # 开启优雅退出保存 config_ck = CheckpointConfig(save_checkpoint_steps=batch_num, keep_checkpoint_max=35, async_save=True, append_info=[{"epoch_num": cur_epoch_num}], exception_save=True) ckpoint_cb = ModelCheckpoint(prefix="train_resnet_cifar10", directory=args_opt.train_url, config=config_ck) # 定义优雅退出ckpt保存callback ckpoint_exp = ExceptionCheckpoint( prefix="train_resnet_cifar10", directory=args_opt.train_url, config=config_ck) # 添加优雅退出ckpt保存callback cb += [ckpoint_cb, ckpoint_exp]

异常处理 Clone仓库失败。可能是网络原因问题。可以在JupyterLab的Terminal中通过执行git clone https://github.com/jupyterlab/extension-examples.git测试网络连通情况。 图6 Clone仓库失败 如果克隆时遇到Notebook当前目录下已有该仓库，系统给出提示仓库名称重复，此时可以单击“覆盖”继续克隆仓库，也可以单击取消。

上传本地超大文件（50GB以上）至JupyterLab 不支持在Notebook的JupyterLab中直接上传大小超过50GB的文件。 50GB以上的文件需要先从本地上传到OBS中，再在Notebook中调用ModelArts的Moxing接口或者SDK接口读写OBS中的文件。 图12 在Notebook中上传下载大文件 具体操作如下： 从本地上传文件至OBS。具体操作请参见上传文件至OBS桶。 将OBS中的文件下载到Notebook，可以通过在Notebook中运行代码的方式完成数据下载，具体方式有2种，ModelArts的SDK接口或者调用MoXing接口。 方法一：使用ModelArts SDK接口将OBS中的文件下载到Notebook后进行操作。 示例代码： 1 2 3 from modelarts.session import Session session = Session() session.obs.copy("obs://bucket-name/obs_file.txt","/home/ma-user/work/") 方法二：使用Moxing操作OBS文件将OBS中的文件同步到Notebook后进行操作。 import moxing as mox #下载一个OBS文件夹sub_dir_0，从OBS下载至Notebook mox.file.copy_parallel('obs://bucket_name/sub_dir_0', '/home/ma-user/work/sub_dir_0') #下载一个OBS文件obs_file.txt，从OBS下载至Notebook mox.file.copy('obs://bucket_name/obs_file.txt', '/home/ma-user/work/obs_file.txt') 如果下载到Notebook中的是zip文件，在Terminal中执行下列命令，解压压缩包。 unzip xxx.zip #在xxx.zip压缩包所在路径直接解压 代码执行完成后，参考图13打开Terminal后执行ls /home/ma-user/work命令查看下载到Notebook中的文件。或者在JupyterLab左侧导航中显示下载的文件，如果没有显示，请刷新后查看，如图14所示。 图13 打开Terminal 图14 查看下载到Notebook中的文件

上传本地大文件（100MB~50GB）至JupyterLab 对于大小超过100MB不超过50GB的文件可以使用OBS中转，系统先将文件上传至OBS（对象桶或并行文件系统），然后从OBS下载到Notebook。下载完成后，ModelArts会将文件自动从OBS中删除。 例如，对于下面这种情况，可以通过“OBS中转”上传。 图5 通过OBS中转上传大文件 如果使用OBS中转需要提供一个OBS中转路径，可以通过以下三种方式提供： 图6 通过OBS中转路径上传 仅第一次单击“OBS中转”需要提供OBS中转路径，以后默认使用该路径直接上传，可以通过上传文件窗口左下角的设置按钮更新OBS中转路径。如图10所示。 方式一：在输入框中直接输入有效的OBS中转路径，然后单击“确定”完成。 图7 输入有效的OBS中转路径 方式二：打开OBS File Browser选择一个OBS中转路径，然后单击“确定”完成。 图8 打开OBS File Browser 方式三：单击“使用默认路径”完成。 图9 使用默认路径上传文件 图10 设置本地文件OBS中转路径 完成OBS中转路径设置后，开始上传文件。 图11 上传文件 解压缩文件包 将文件以压缩包形式上传至Notebook JupyterLab后， 可在Terminal中解压缩文件包。 unzip xxx.zip #在xxx.zip压缩包所在路径直接解压 解压命令的更多使用说明可以在主流搜索引擎中查找Linux解压命令操作。

上传文件要求 对于大小不超过100MB的文件直接上传，并展示文件大小、上传进度及速度等详细信息。 对于大小超过100MB不超过50GB的文件可以使用OBS中转，系统先将文件上传OBS（对象桶或并行文件系统），然后从OBS下载到Notebook，上传完成后，会将文件从OBS中删除。 50GB以上的文件上传通过调用ModelArts SDK或者Moxing完成。 对于Notebook当前目录下已经有同文件名称的文件，可以覆盖继续上传，也可以取消。 支持10个文件同时上传，其余文件显示“等待上传”。不支持上传文件夹，可以将文件夹压缩成压缩包上传至Notebook后， 在Terminal中解压压缩包。 unzip xxx.zip #在xxx.zip压缩包所在路径直接解压 解压命令的更多使用说明可以在主流搜索引擎中查找Linux解压命令操作。 多个文件同时上传时，JupyterLab窗口最下面会显示上传文件总数和已上传文件数。