华为云用户手册

在Workflow中指定仅运行部分节点 Workflow通过支持预置场景的方式来实现部分运行的能力，在开发工作流时按照场景的不同对DAG进行划分，之后在运行态可选择任意场景单独运行。具体代码示例如下所示： workflow =wf.Workflow( name="image_cls", desc="this is a demo workflow", steps=[label_step, release_data_step, training_step, model_step, service_step], policy=wf.policy.Policy( scenes=[ wf.policy.Scene( scene_name="模型训练", scene_steps=[label_step, release_data_step, training_step] ), wf.policy.Scene( scene_name="服务部署", scene_steps=[model_step, service_step] ), ] ) ) 该示例中Workflow包含了五个节点（节点相关定义已省略），在policy中定义了两个预置场景：模型训练和服务部署，工作流发布至运行态后，部分运行的开关默认关闭，节点全部运行。用户可在权限管理页面打开开关，选择指定的场景进行运行。 部分运行能力支持同一个节点被定义在不同的运行场景中，但是需要用户自行保证节点之间数据依赖的正确性。另外，部分运行能力仅支持在运行态进行配置运行，不支持在开发态进行调试。 父主题： Workflow高阶能力

发布Workflow至运行态并运行 该方式支持用户直接在SDK侧发布并运行工作流，节省了前往控制台进行配置运行的操作，对Workflow代码改造如下。 from modelarts import workflow as wf # 定义统一存储对象管理输出目录 output_storage = wf.data.OutputStorage(name="output_storage", description="输出目录统一配置", default="**") # 数据集对象 dataset = wf.data.DatasetPlaceholder(name="input_data", default=wf.data.Dataset(dataset_name="**", version_name="**")) # 创建训练作业 job_step = wf.steps.JobStep( name="training_job", title="图像分类训练", algorithm=wf.AIGalleryAlgorithm( subscription_id="**", # 图像分类算法的订阅ID，自行前往算法管理页面进行查看，可选参数，此处以订阅算法举例 item_version_id="10.0.0", # 订阅算法的版本号，可选参数，此处以订阅算法举例 parameters=[ wf.AlgorithmParameters(name="task_type", value="image_classification_v2"), wf.AlgorithmParameters(name="model_name", value="resnet_v1_50"), wf.AlgorithmParameters(name="do_train", value="True"), wf.AlgorithmParameters(name="do_eval_along_train", value="True"), wf.AlgorithmParameters(name="variable_update", value="horovod"), wf.AlgorithmParameters(name="learning_rate_strategy", value=wf.Placeholder(name="learning_rate_strategy", placeholder_type=wf.PlaceholderType.STR, default="0.002", description="训练的学习率策略(10:0.001,20:0.0001代表0-10个epoch学习率0.001，10-20epoch学习率0.0001),如果不指定epoch, 会根据验证精度情况自动调整学习率，并当精度没有明显提升时，训练停止")), wf.AlgorithmParameters(name="batch_size", value=wf.Placeholder(name="batch_size", placeholder_type=wf.PlaceholderType.INT, default=64, description="每步训练的图片数量（单卡）")), wf.AlgorithmParameters(name="eval_batch_size", value=wf.Placeholder(name="eval_batch_size", placeholder_type=wf.PlaceholderType.INT, default=64, description="每步验证的图片数量（单卡）")), wf.AlgorithmParameters(name="evaluate_every_n_epochs", value=wf.Placeholder(name="evaluate_every_n_epochs", placeholder_type=wf.PlaceholderType.FLOAT, default=1.0, description="每训练n个epoch做一次验证")), wf.AlgorithmParameters(name="save_model_secs", value=wf.Placeholder(name="save_model_secs", placeholder_type=wf.PlaceholderType.INT, default=60, description="保存模型的频率（单位：s）")), wf.AlgorithmParameters(name="save_summary_steps", value=wf.Placeholder(name="save_summary_steps", placeholder_type=wf.PlaceholderType.INT, default=10, description="保存summary的频率（单位：步）")), wf.AlgorithmParameters(name="log_every_n_steps", value=wf.Placeholder(name="log_every_n_steps", placeholder_type=wf.PlaceholderType.INT, default=10, description="打印日志的频率（单位：步）")), wf.AlgorithmParameters(name="do_data_cleaning", value=wf.Placeholder(name="do_data_cleaning", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否进行数据清洗, 数据格式异常会导致训练失败，建议开启，保证训练稳定性。数据量过大时，数据清洗可能耗时较久，可自行线下清洗（支持BMP.JPEG,PNG格式, RGB三通道）。建议用JPEG格式数据")), wf.AlgorithmParameters(name="use_fp16", value=wf.Placeholder(name="use_fp16", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否使用混合精度, 混合精度可以加速训练，但是可能会造成一点精度损失，如果对精度无极严格的要求，建议开启")), wf.AlgorithmParameters(name="xla_compile", value=wf.Placeholder(name="xla_compile", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否开启xla编译，加速训练，默认启用")), wf.AlgorithmParameters(name="data_format", value=wf.Placeholder(name="data_format", placeholder_type=wf.PlaceholderType.ENUM, default="NCHW", enum_list=["NCHW", "NHWC"], description="输入数据类型，NHWC表示channel在最后，NCHW表channel在最前，默认值NCHW（速度有提升）")), wf.AlgorithmParameters(name="best_model", value=wf.Placeholder(name="best_model", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否在训练过程中保存并使用精度最高的模型，而不是最新的模型。默认值True,保存最优模型。在一定误差范围内，最优模型会保存最新的高精度模型")), wf.AlgorithmParameters(name="jpeg_preprocess", value=wf.Placeholder(name="jpeg_preprocess", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否使用jpeg预处理加速算子(仅支持jpeg格式数据)，可加速数据读取，提升性能，默认启用。如果数据格式不是jpeg格式，开启数据清洗功能即可使用")) ] ), inputs=[wf.steps.JobInput(name="data_url", data=dataset)], outputs=[wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=output_storage.join("/train_output/")))], spec=wf.steps.JobSpec( resource=wf.steps.JobResource( flavor=wf.Placeholder( name="training_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格", default={"flavor_id": "**"} ) ) ) ) # 构建工作流对象 workflow = wf.Workflow( name="image-classification-ResNeSt", desc="this is an image classification workflow", steps=[job_step], storages=[output_storage] ) 用户需要完成上述代码中**部分的配置，主要涉及以下三项。 统一存储：output_storage对象的default值，需填写一个已存在的OBS路径，路径格式为：/OBS桶名称/文件夹路径/。 数据集对象：使用准备数据集章节下载的数据集即可，填写相应的数据集名称以及版本号。 训练资源规格：配置计算资源。由于举例的算法只能跑GPU，此处必须配置GPU类型的资源，可使用免费规格（modelarts.p3.large.public.free）。 配置项修改完成后执行如下代码。 workflow.release_and_run() 执行完成后可前往ModelArts管理控制台，在总览页中选择Workflow，查看工作流的运行情况。

发布Workflow至运行态 工作流编写完成后，可以进行固化保存，调用Workflow对象的release()方法发布到运行态进行配置执行（在管理控制台Workflow页面配置）。 执行如下命令： workflow.release() 上述命令执行完成后，如果日志打印显示发布成功，则可前往ModelArts的Workflow页面中查看新发布的工作流，进入Workflow详情，单击“配置”进行参数配置。工作流相关的配置执行操作可参考如何使用Workflow。 基于release()方法，提供了release_and_run()方法，支持用户在开发态发布并运行工作流，节省了前往console配置执行的操作。 使用该方法时需要注意以下几个事项： Workflow中所有出现占位符相关的配置对象时，均需要设置默认值，或者直接使用固定的数据对象 方法的执行依赖于Workflow对象的名称：当该名称的工作流不存在时，则创建新工作流并创建新执行；当该名称的工作流已存在时，则更新存在的工作流并基于新的工作流结构创建新的执行 workflow.release_and_run()

导入Workflow Data包 在编写Workflow过程中，相关对象都通过Workflow包进行导入，梳理如下： from modelarts import workflow as wf Data包相关内容导入： wf.data.DatasetTypeEnum wf.data.Dataset wf.data.DatasetVersionConfig wf.data.DatasetPlaceholder wf.data.ServiceInputPlaceholder wf.data.ServiceData wf.data.ServiceUpdatePlaceholder wf.data.DataTypeEnum wf.data.ModelData wf.data.GalleryModel wf.data.OBSPath wf.data.OBSOutputConfig wf.data.OBSPlaceholder wf.data.SWRImage wf.data.SWRImagePlaceholder wf.data.Storage wf.data.InputStorage wf.data.OutputStorage wf.data.LabelTask wf.data.LabelTaskPlaceholder wf.data.LabelTaskConfig wf.data.LabelTaskTypeEnum wf.data.MetricsConfig wf.data.TripartiteServiceConfig wf.data.DataConsumptionSelector policy包相关内容导入： wf.policy.Policy wf.policy.Scene steps包相关内容导入： wf.steps.MetricInfo wf.steps.Condition wf.steps.ConditionTypeEnum wf.steps.ConditionStep wf.steps.LabelingStep wf.steps.LabelingInput wf.steps.LabelingOutput wf.steps.LabelTaskProperties wf.steps.ImportDataInfo wf.steps.DataOriginTypeEnum wf.steps.DatasetImportStep wf.steps.DatasetImportInput wf.steps.DatasetImportOutput wf.steps.AnnotationFormatConfig wf.steps.AnnotationFormatParameters wf.steps.AnnotationFormatEnum wf.steps.Label wf.steps.ImportTypeEnum wf.steps.LabelFormat wf.steps.LabelTypeEnum wf.steps.ReleaseDatasetStep wf.steps.ReleaseDatasetInput wf.steps.ReleaseDatasetOutput wf.steps.CreateDatasetStep wf.steps.CreateDatasetInput wf.steps.CreateDatasetOutput wf.steps.DatasetProperties wf.steps.SchemaField wf.steps.ImportConfig wf.steps.JobStep wf.steps.JobMetadata wf.steps.JobSpec wf.steps.JobResource wf.steps.JobTypeEnum wf.steps.JobEngine wf.steps.JobInput wf.steps.JobOutput wf.steps.LogExportPath wf.steps.MrsJobStep wf.steps.MrsJobInput wf.steps.MrsJobOutput wf.steps.MrsJobAlgorithm wf.steps.ModelStep wf.steps.ModelInput wf.steps.ModelOutput wf.steps.ModelConfig wf.steps.Template wf.steps.TemplateInputs wf.steps.ServiceStep wf.steps.ServiceInput wf.steps.ServiceOutput wf.steps.ServiceConfig wf.steps.StepPolicy Workflow包相关内容导入： wf.workflow wf.Subgraph wf.Placeholder wf.PlaceholderType wf.AlgorithmParameters wf.BaseAlgorithm wf.Algorithm wf.AIGalleryAlgorithm wf.resource wf.SystemEnv wf.add_whitelist_users wf.delete_whitelist_users

编写工作流代码示例 以图像分类为例，阐述机器学习端到端场景的完整开发过程，主要包括数据标注、模型训练、服务部署等过程。您需要准备如下算法和数据集。 准备一个图像分类算法（或者可以直接从AI Gallery搜索订阅一个“图像分类-ResNet_v1_50”算法）。 准备一个图片类型的数据集，请参考准备数据集。可从AI Gallery直接下载（例如：8类常见生活垃圾图片数据集）。 from modelarts import workflow as wf # 定义统一存储对象管理输出目录 output_storage = wf.data.OutputStorage(name="output_storage", description="输出目录统一配置") # 创建标注任务 data = wf.data.DatasetPlaceholder(name="input_data") label_step = wf.steps.LabelingStep( name="labeling", title="数据标注", properties=wf.steps.LabelTaskProperties( task_type=wf.data.LabelTaskTypeEnum.IMAGE_CLASSIFICATION, task_name=wf.Placeholder(name="task_name", placeholder_type=wf.PlaceholderType.STR, description="请输入一个只包含大小写字母、数字、下划线、中划线或者中文字符的名称。填写已有标注任务名称，则直接使用该标注任务；填写新标注任务名称，则自动创建新的标注任务") ), inputs=wf.steps.LabelingInput(name="labeling_input", data=data), outputs=wf.steps.LabelingOutput(name="labeling_output"), ) # 对标注任务进行发布 release_step = wf.steps.ReleaseDatasetStep( name="release", title="数据集版本发布", inputs=wf.steps.ReleaseDatasetInput(name="input_data", data=label_step.outputs["labeling_output"].as_input()), outputs=wf.steps.ReleaseDatasetOutput(name="labeling_output", dataset_version_config=wf.data.DatasetVersionConfig(train_evaluate_sample_ratio="0.8")), depend_steps=[label_step] ) # 创建训练作业 job_step = wf.steps.JobStep( name="training_job", title="图像分类训练", algorithm=wf.AIGalleryAlgorithm( subscription_id="***", # 订阅算法的ID，自行补充 item_version_id="10.0.0", # 订阅算法的版本ID parameters=[ wf.AlgorithmParameters(name="task_type", value="image_classification_v2"), wf.AlgorithmParameters(name="model_name", value="resnet_v1_50"), wf.AlgorithmParameters(name="do_train", value="True"), wf.AlgorithmParameters(name="do_eval_along_train", value="True"), wf.AlgorithmParameters(name="variable_update", value="horovod"), wf.AlgorithmParameters(name="learning_rate_strategy", value=wf.Placeholder(name="learning_rate_strategy", placeholder_type=wf.PlaceholderType.STR, default="0.002", description="训练的学习率策略(10:0.001,20:0.0001代表0-10个epoch学习率0.001，10-20epoch学习率0.0001),如果不指定epoch, 会根据验证精度情况自动调整学习率，并当精度没有明显提升时，训练停止")), wf.AlgorithmParameters(name="batch_size", value=wf.Placeholder(name="batch_size", placeholder_type=wf.PlaceholderType.INT, default=64, description="每步训练的图片数量（单卡）")), wf.AlgorithmParameters(name="eval_batch_size", value=wf.Placeholder(name="eval_batch_size", placeholder_type=wf.PlaceholderType.INT, default=64, description="每步验证的图片数量（单卡）")), wf.AlgorithmParameters(name="evaluate_every_n_epochs", value=wf.Placeholder(name="evaluate_every_n_epochs", placeholder_type=wf.PlaceholderType.FLOAT, default=1.0, description="每训练n个epoch做一次验证")), wf.AlgorithmParameters(name="save_model_secs", value=wf.Placeholder(name="save_model_secs", placeholder_type=wf.PlaceholderType.INT, default=60, description="保存模型的频率（单位：s）")), wf.AlgorithmParameters(name="save_summary_steps", value=wf.Placeholder(name="save_summary_steps", placeholder_type=wf.PlaceholderType.INT, default=10, description="保存summary的频率（单位：步）")), wf.AlgorithmParameters(name="log_every_n_steps", value=wf.Placeholder(name="log_every_n_steps", placeholder_type=wf.PlaceholderType.INT, default=10, description="打印日志的频率（单位：步）")), wf.AlgorithmParameters(name="do_data_cleaning", value=wf.Placeholder(name="do_data_cleaning", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否进行数据清洗, 数据格式异常会导致训练失败，建议开启，保证训练稳定性。数据量过大时，数据清洗可能耗时较久，可自行线下清洗（支持BMP.JPEG,PNG格式, RGB三通道）。建议用JPEG格式数据")), wf.AlgorithmParameters(name="use_fp16", value=wf.Placeholder(name="use_fp16", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否使用混合精度, 混合精度可以加速训练，但是可能会造成一点精度损失，如果对精度无极严格的要求，建议开启")), wf.AlgorithmParameters(name="xla_compile", value=wf.Placeholder(name="xla_compile", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否开启xla编译，加速训练，默认启用")), wf.AlgorithmParameters(name="data_format", value=wf.Placeholder(name="data_format", placeholder_type=wf.PlaceholderType.ENUM, default="NCHW", enum_list=["NCHW", "NHWC"], description="输入数据类型，NHWC表示channel在最后，NCHW表channel在最前，默认值NCHW（速度有提升）")), wf.AlgorithmParameters(name="best_model", value=wf.Placeholder(name="best_model", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否在训练过程中保存并使用精度最高的模型，而不是最新的模型。默认值True,保存最优模型。在一定误差范围内，最优模型会保存最新的高精度模型")), wf.AlgorithmParameters(name="jpeg_preprocess", value=wf.Placeholder(name="jpeg_preprocess", placeholder_type=wf.PlaceholderType.STR, default="True", description="是否使用jpeg预处理加速算子(仅支持jpeg格式数据)，可加速数据读取，提升性能，默认启用。如果数据格式不是jpeg格式，开启数据清洗功能即可使用")) ] ), inputs=[wf.steps.JobInput(name="data_url", data=release_step.outputs["labeling_output"].as_input())], outputs=[wf.steps.JobOutput(name="train_url", obs_config=wf.data.OBSOutputConfig(obs_path=output_storage.join("/train_output/")))], spec=wf.steps.JobSpec( resource=wf.steps.JobResource( flavor=wf.Placeholder(name="training_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格" ) ) ), depend_steps=[release_step] ) model_name = wf.Placeholder(name="model_name", placeholder_type=wf.PlaceholderType.STR, description="请输入一个1至64位且只包含大小写字母、中文、数字、中划线或者下划线的名称。工作流第一次运行建议填写新的模型名称，后续运行会自动在该模型上新增版本") # 模型注册 model_step = wf.steps.ModelStep( name="model_step", title="模型注册", inputs=[wf.steps.ModelInput(name="model_input", data=job_step.outputs["train_url"].as_input())], outputs=[wf.steps.ModelOutput(name="model_output", model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow"))], depend_steps=[job_step] ) # 服务部署 service_step = wf.steps.ServiceStep( name="service_step", title="服务部署", inputs=[wf.steps.ServiceInput(name="service_input", data=wf.data.ServiceInputPlaceholder(name="service_model", model_name=model_name))], outputs=[wf.steps.ServiceOutput(name="service_output")], depend_steps=[model_step] ) # 构建工作流对象 workflow = wf.Workflow(name="image-classification-ResNeSt", desc="this is a image classification workflow", steps=[label_step, release_step, job_step, model_step, service_step], storages=[output_storage] ) 在工作流编写完成后可自行进行发布等操作。

使用案例 from modelarts import workflow as wf condition_equal = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.EQ, left=wf.Placeholder(name="is_true", placeholder_type=wf.PlaceholderType.BOOL), right=True) condition_step = wf.steps.ConditionStep( name="condition_step", conditions=[condition_equal], if_then_steps=["training_job_1"], else_then_steps=["training_job_2"], ) # 构建一个OutputStorage对象，对训练输出目录做统一管理 storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 # 定义输入的OBS对象 obs_data = wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory") # 通过JobStep来定义一个训练节点，并将训练结果输出到OBS job_step_1 = wf.steps.JobStep( name="training_job_1", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="图像分类训练", # 标题信息，不填默认使用name algorithm=wf.AIGalleryAlgorithm( subscription_id="subscription_id", # 算法订阅ID item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号 parameters=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 inputs=wf.steps.JobInput(name="data_url", data=obs_data), # JobStep的输入在运行时配置；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="训练资源规格") ) ), # 训练资源规格信息 depend_steps=[condition_step] ) # 通过JobStep来定义一个训练节点，并将训练结果输出到OBS job_step_2 = wf.steps.JobStep( name="training_job_2", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="图像分类训练", # 标题信息，不填默认使用name algorithm=wf.AIGalleryAlgorithm( subscription_id="subscription_id", # 算法订阅ID item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号 parameters=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 inputs=wf.steps.JobInput(name="data_url", data=obs_data), # JobStep的输入在运行时配置；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="训练资源规格") ) ), # 训练资源规格信息 depend_steps=[condition_step] ) # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_step = wf.steps.ModelStep( name="model_registration", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 inputs=wf.steps.ModelInput(name='model_input', data=wf.data.DataConsumptionSelector(data_list=[job_step_1.outputs["train_url"].as_input(), job_step_2.outputs["train_url"].as_input()])), # 选择job_step_1或者job_step_2的输出作为输入 outputs=wf.steps.ModelOutput(name='model_output', model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow")), # ModelStep的输出 depend_steps=[job_step_1, job_step_2] # 依赖的作业类型节点对象 )# job_step是wf.steps.JobStep的 实例对象，train_url是wf.steps.JobOutput的name字段值 workflow = wf.Workflow(name="data-select-demo", desc="this is a test workflow", steps=[condition_step, job_step_1, job_step_2, model_step], storages=storage ) 案例中的Workflow存在两个并行分支，并且同时只有一条分支会执行，由condition_step的相关配置决定。model_step的输入来源为job_step_1或者job_step_2的输出，当job_step_1节点所在分支执行，job_step_2节点所在分支跳过时，model_step节点执行时自动获取job_step_1的输出作为输入，反之自动获取job_step_2的输出作为输入。

控制多分支的部分执行 from modelarts import workflow as wf # 构建一个OutputStorage对象，对训练输出目录做统一管理 storage = wf.data.Storage(name="storage_name", title="title_info", with_execution_id=True, create_dir=True, description="description_info") # name字段必填，title, description可选填 # 定义输入的OBS对象 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来定义一个训练节点，并将训练结果输出到OBS job_step_a = wf.steps.JobStep( name="training_job_a", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="图像分类训练", # 标题信息，不填默认使用name algorithm=wf.AIGalleryAlgorithm( subscription_id="subscription_id", # 算法订阅ID item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号 parameters=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 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="训练资源规格") ) ), # 训练资源规格信息 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来定义一个训练节点，并将训练结果输出到OBS job_step_b = wf.steps.JobStep( name="training_job_b", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="图像分类训练", # 标题信息，不填默认使用name algorithm=wf.AIGalleryAlgorithm( subscription_id="subscription_id", # 算法订阅ID item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号 parameters=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 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="训练资源规格") ) ), # 训练资源规格信息 policy=wf.steps.StepPolicy(skip_conditions=condition_equal_b) ) # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_step = wf.steps.ModelStep( name="model_registration", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 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], # 依赖的作业类型节点对象 ) workflow = wf.Workflow( name="new-condition-demo", desc="this is a demo workflow", steps=[job_step_a, job_step_b, model_step], storages=storage )

控制单节点的执行 通过参数配置实现 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) # 构建一个OutputStorage对象，对训练输出目录做统一管理 storage = wf.data.OutputStorage(name="storage_name", title="title_info", description="description_info") # name字段必填，title, description可选填 # 定义输入的OBS对象 obs_data = wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory") # 通过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=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 inputs=wf.steps.JobInput(name="data_url", data=obs_data), # JobStep的输入在运行时配置；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="训练资源规格") ) ), # 训练资源规格信息 policy=wf.steps.StepPolicy( skip_conditions=[condition_equal] # 通过skip_conditions中的计算结果决定job_step是否跳过 ) ) workflow = wf.Workflow( name="new-condition-demo", desc="this is a demo workflow", steps=[job_step], storages=storage ) 案例中job_step配置了相关的跳过策略，并且通过一个bool类型的参数进行控制。当name为is_skip的Placeholder参数配置为True时，condition_equal的计算结果为True，此时job_step会被置为跳过，反之job_step正常执行，其中Condition对象详情可参考构建条件节点控制分支执行。 通过获取JobStep输出的相关metric指标信息实现 from modelarts import workflow as wf # 构建一个OutputStorage对象，对训练输出目录做统一管理 storage = wf.data.Storage(name="storage_name", title="title_info", with_execution_id=True, create_dir=True, description="description_info") # name字段必填，title, description可选填 # 定义输入的OBS对象 obs_data = wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory") # 通过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=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 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的输出路径，相关指标信息由作业脚本代码根据指定的数据格式自行输出（示例中需要将metric信息输出到训练输出目录下的metrics.json文件中） ], spec=wf.steps.JobSpec( resource=wf.steps.JobResource( flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格") ) ) # 训练资源规格信息 ) # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) # 定义条件对象 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", # 模型注册节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="模型注册", # 标题信息 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], # 依赖的作业类型节点对象 policy=wf.steps.StepPolicy(skip_conditions=condition_lt) # 通过skip_conditions中的计算结果决定model_step是否跳过 ) workflow = wf.Workflow( name="new-condition-demo", desc="this is a demo workflow", steps=[job_step, model_step], storages=storage ) 案例中model_step配置了相关的跳过策略，并且通过获取job_step输出的accuracy指标信息与预置的值进行比较，决定是否需要进行模型注册。当job_step输出的accuracy指标数据小于阈值0.5时，condition_lt的计算结果为True，此时model_step会被置为跳过，反之model_step正常执行。 job_step输出的metric文件格式要求可参考创建Workflow训练作业节点部分，并且在Condition中只支持使用type为float类型的指标数据作为输入。 此案例中metrics.json的内容示例如下： [ { "key": "loss", // 指标数据名称，不支持特殊字符，长度限制为64字符 "title": "loss", // 指标数据标题，长度限制为64字符 "type": "float", // 指标数据类型，支持以下类型：浮点：float、折线图：line chart、柱状图：histogram、矩阵：table、一维表格：one-dimensional-table "data": { "value": 1.2 // 指标数据值，不同类型的使用示例可以参考创建Workflow训练作业节点 } }, { "key": "accuracy", "title": "accuracy", "type": "float", "data": { "value": 0.8 } } ]

Workflow多分支运行介绍 当前支持两种方式实现多分支的能力，条件节点只支持双分支的选择执行，局限性较大，推荐使用配置节点参数控制分支执行的方式，可以在不添加新节点的情况下完全覆盖ConditionStep的能力，使用上更灵活。 构建条件节点控制分支执行主要用于执行流程的条件分支选择，可以简单的进行数值比较来控制执行流程，也可以根据节点输出的metric相关信息决定后续的执行流程。 配置节点参数控制分支执行与ConditionStep的使用场景类似，但功能更加强大。主要用于存在多分支选择执行的复杂场景，在每次启动执行后需要根据相关配置信息决定哪些分支需要执行，哪些分支需要跳过，达到分支部分执行的目的。 父主题： 构建Workflow多分支运行场景

进阶示例 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的能力，详情请参见配置节点参数控制分支执行章节。

简单示例 通过参数配置实现 import modelarts.workflow as wf left_value = wf.Placeholder(name="left_value", placeholder_type=wf.PlaceholderType.BOOL, default=True) # 条件对象 condition = wf.steps.Condition(condition_type=wf.steps.ConditionTypeEnum.EQ, left=left_value, right=True) # 条件对象，包含类型以及左右值 # 条件节点 condition_step = wf.steps.ConditionStep( name="condition_step_test", # 条件节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 conditions=condition, # 条件对象,允许多个条件，条件之间的关系为&& 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 ) # 该节点仅作为示例使用，其他字段需自行补充 model_step_1 = wf.steps.ModelStep( name="model_step_1", depend_steps=job_step_1 ) # 该节点仅作为示例使用，其他字段需自行补充 job_step_2 = wf.steps.JobStep( name="job_step_2", depend_steps=condition_step ) # 该节点仅作为示例使用，其他字段需自行补充 model_step_2 = wf.steps.ModelStep( name="model_step_2", depend_steps=job_step_2 ) workflow = wf.Workflow( name="condition-demo", desc="this is a demo workflow", steps=[condition_step, job_step_1, job_step_2, model_step_1, model_step_2] ) 场景说明：job_step_1和job_step_2表示两个训练节点，并且均直接依赖于condition_step。condition_step通过参数配置决定后继节点的执行行为。 执行情况分析： 参数left_value默认值为True，则condition逻辑表达式计算结果为True：job_step_1执行，job_step_2跳过，并且以job_step_2为唯一根节点的分支所包含的所有节点也将跳过，即model_step_2会跳过，因此最终执行的节点有condition_step、job_step_1、model_step_1。 如果设置left_value的值为Fasle，则condition逻辑表达式计算结果为False：job_step_2执行，job_step_1跳过，并且以job_step_1为唯一根节点的分支所包含的所有节点也将跳过，即model_step_1会跳过，因此最终执行的节点有condition_step、job_step_2、model_step_2。 通过获取JobStep输出的相关metric指标信息实现 from modelarts import workflow as wf # 构建一个OutputStorage对象，对训练输出目录做统一管理 storage = wf.data.Storage(name="storage_name", title="title_info", with_execution_id=True, create_dir=True, description="description_info") # name字段必填，title, description可选填 # 定义输入的OBS对象 obs_data = wf.data.OBSPlaceholder(name="obs_placeholder_name", object_type="directory") # 通过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=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 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的输出路径，相关指标信息由作业脚本代码根据指定的数据格式自行输出（示例中需要将metric信息输出到训练输出目录下的metrics.json文件中） ], spec=wf.steps.JobSpec( resource=wf.steps.JobResource( flavor=wf.Placeholder(name="train_flavor", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格") ) ) # 训练资源规格信息 ) # 定义条件对象 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 ) condition_step = wf.steps.ConditionStep( name="condition_step_test", # 条件节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 conditions=condition_lt, # 条件对象,允许多个条件，条件之间的关系为&& if_then_steps="training_job_retrain", # 当condition结果为true时，名称为training_job_retrain的节点允许执行，名称为model_registration的节点跳过不执行 else_then_steps="model_registration", # 当condition结果为false时，名称为model_registration的节点允许执行，名称为training_job_retrain的节点跳过不执行 depend_steps=job_step ) # 通过JobStep来定义一个训练节点，并将训练结果输出到OBS job_step_retrain = wf.steps.JobStep( name="training_job_retrain", # 训练节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="图像分类重新训练训练", # 标题信息，不填默认使用name algorithm=wf.AIGalleryAlgorithm( subscription_id="subscription_id", # 算法订阅ID item_version_id="item_version_id", # 算法订阅版本ID，也可直接填写版本号 parameters=[] ), # 训练使用的算法对象，示例中使用AIGallery订阅的算法；部分算法超参的值如果无需修改，则在parameters字段中可以不填写，系统自动填充相关超参值 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_retrain"))), wf.steps.JobOutput(name="metrics", metrics_config=wf.data.MetricsConfig(metric_files=storage.join("directory_path_retrain/metrics.json", create_dir=False))) # 指定metric的输出路径，相关指标信息由作业脚本代码根据指定的数据格式自行输出（示例中需要将metric信息输出到训练输出目录下的metrics.json文件中） ], spec=wf.steps.JobSpec( resource=wf.steps.JobResource( flavor=wf.Placeholder(name="train_flavor_retrain", placeholder_type=wf.PlaceholderType.JSON, description="训练资源规格") ) ), # 训练资源规格信息 depend_steps=condition_step ) # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) model_step = 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()), # job_step的输出作为输入 outputs=wf.steps.ModelOutput(name='model_output', model_config=wf.steps.ModelConfig(model_name=model_name, model_type="TensorFlow")), # ModelStep的输出 depend_steps=condition_step, ) workflow = wf.Workflow( name="condition-demo", desc="this is a demo workflow", steps=[job_step, condition_step, job_step_retrain, model_step], storages=storage ) 案例中ConditionStep节点通过获取job_step输出的accuracy指标信息与预置的值进行比较，决定重新训练还是模型注册。当job_step输出的accuracy指标数据小于阈值0.5时，condition_lt的计算结果为True，此时job_step_retrain运行，model_step跳过；反之job_step_retrain跳过，model_step执行。 job_step输出的metric文件格式要求可参考创建Workflow训练作业节点部分，并且在Condition中只支持使用type为float类型的指标数据作为输入。 此案例中metrics.json的内容示例如下： [ { "key": "loss", "title": "loss", "type": "float", "data": { "value": 1.2 } }, { "key": "accuracy", "title": "accuracy", "type": "float", "data": { "value": 0.8 } } ]

新增在线服务 import modelarts.workflow as wf # 通过ServiceStep来定义一个服务部署节点，输入指定的模型进行服务部署 # 定义模型名称参数 model_name = wf.Placeholder(name="placeholder_name", placeholder_type=wf.PlaceholderType.STR) service_step = wf.steps.ServiceStep( name="service_step", # 服务部署节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 title="新增服务", # 标题信息 inputs=wf.steps.ServiceInput(name="si_service_ph", data=wf.data.ServiceInputPlaceholder(name="si_placeholder1", # 模型名称的限制/约束,在运行态只能选择该模型名称；一般与模型注册节点中的model_name使用同一个参数对象 model_name=model_name)),# ServiceStep的输入列表 outputs=wf.steps.ServiceOutput(name="service_output") # ServiceStep的输出 ) workflow = wf.Workflow( name="service-step-demo", desc="this is a demo workflow", steps=[service_step] )

属性总览 您可以使用ServiceStep来构建服务部署节点，ServiceStep结构如下 表1 ServiceStep 属性 描述 是否必填 数据类型 name 服务部署节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 是 str inputs 服务部署节点的输入列表 否 ServiceInput或者ServiceInput的列表 outputs 服务部署节点的输出列表 是 ServiceOutput或者ServiceOutput的列表 title title信息，主要用于前端的名称展示 否 str description 服务部署节点的描述信息 否 str policy 节点执行的policy 否 StepPolicy depend_steps 依赖的节点列表 否 Step或者Step的列表 表2 ServiceInput 属性 描述 是否必填 数据类型 name 服务部署节点的输入名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输入名称不能重复 是 str data 服务部署节点的输入数据对象 是 模型列表或服务相关对象，当前仅支持ServiceInputPlaceholder，ServiceData，ServiceUpdatePlaceholder 表3 ServiceOutput 属性 描述 是否必填 数据类型 name 服务部署节点的输出名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输出名称不能重复 是 str service_config 服务部署相关配置信息 是 ServiceConfig 表4 ServiceConfig 属性 描述 是否必填 数据类型 infer_type 推理方式：取值可为real-time/batch/edge。默认为real-time。 real-time代表在线服务，将模型部署为一个Web Service。 batch为批量服务，批量服务可对批量数据进行推理，完成数据处理后自动停止。 edge表示边缘服务，通过华为云智能边缘平台，在边缘节点将模型部署为一个Web Service，需提前在IEF（智能边缘服务）创建好节点。 是 str service_name 服务名称，支持1-64位可见字符（含中文），名称可以包含字母、中文、数字、中划线、下划线。 说明： 该字段不填时默认为自动生成的服务名称。 否 str、Placeholder description 服务备注，默认为空，不超过100个字符。 否 str vpc_id 在线服务实例部署的虚拟私有云ID，默认为空，此时ModelArts会为每个用户分配一个专属的VPC，用户之间隔离。如需要在服务实例中访问名下VPC内的其他服务组件，则可配置此参数为对应VPC的ID。VPC一旦配置，不支持修改。当vpc_id与cluster_id一同配置时，只有专属资源池参数生效。 否 str subnet_network_id 子网的网络ID，默认为空，当配置了vpc_id则此参数必填。需填写虚拟私有云控制台子网详情中显示的“网络ID”。通过子网可提供与其他网络隔离的、可以独享的网络资源。 否 str security_group_id 安全组，默认为空，当配置了vpc_id则此参数必填。安全组起着虚拟防火墙的作用，为服务实例提供安全的网络访问控制策略。安全组须包含至少一条入方向规则，对协议为TCP、源地址为0.0.0.0/0、端口为8080的请求放行。 否 str cluster_id 专属资源池ID，默认为空，不使用专属资源池。使用专属资源池部署服务时需确保集群状态正常；配置此参数后，则使用集群的网络配置，vpc_id参数不生效；与下方real-time config中的cluster_id同时配置时，优先使用real-time config中的cluster_id参数。 否 str additional_properties 附加的相关配置信息。 否 dict apps 服务部署支持APP认证。支持填入多个app name。 否 str、Placeholder、list envs 环境变量 否 dict 示例： example = ServiceConfig() # 主要在服务部署节点的输出中使用 如果您没有特殊需求，可直接使用内置的默认值。

从自定义镜像中注册模型 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的方式注册模型 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] )

从训练作业中注册模型（模型输入来源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] )

从训练作业中注册模型（模型输入来源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] )

资源规格查询 您在创建作业类型节点之前可以通过以下操作来获取该账号所支持的训练资源规格列表以及引擎规格列表： 导包 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对象进行配置。 表16 MetricsConfig 属性 描述 是否必填 数据类型 metric_files metrics输出文件列表 是 list，列表内元素支持（str、Placeholder、Storage） realtime_visualization 输出的metrics信息是否需要实时展示 否 bool，默认为False visualization 是否呈现独立的可视化节点 否 bool，默认为True 对于输出的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。

属性总览 您可以使用JobStep来构建作业类型节点，JobStep结构如下 表1 JobStep 属性 描述 是否必填 数据类型 name 作业节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复 是 str algorithm 算法对象 是 BaseAlgorithm Algorithm AIGalleryAlgorithm spec 作业使用的资源规格相关配置 是 JobSpec inputs 作业节点的输入列表 是 JobInput或者JobInput的列表 outputs 作业节点的输出列表 是 JobOutput或者JobOutput的列表 title title信息，主要用于前端的名称展示 否 str description 作业节点的描述信息 否 str policy 节点执行的policy 否 StepPolicy depend_steps 依赖的节点列表 否 Step或者Step的列表 表2 JobInput 属性 描述 是否必填 数据类型 name 作业类型节点的输入名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输入名称不能重复 是 str data 作业类型节点的输入数据对象 是 数据集或OBS相关对象，当前仅支持Dataset、DatasetPlaceholder、DatasetConsumption、OBSPath、OBSConsumption、OBSPlaceholder、DataConsumptionSelector 表3 JobOutput 属性 描述 是否必填 数据类型 name 作业类型节点的输出名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输出名称不能重复 是 str obs_config 输出的OBS相关配置 否 OBSOutputConfig model_config 输出的模型相关配置 否 ModelConfig metrics_config metrics相关配置 否 MetricsConfig 表4 OBSOutputConfig 属性 描述 是否必填 数据类型 obs_path 已存在的OBS目录 是 str、Placeholder、Storage metric_file 存储metric信息的文件名称 否 str、Placeholder 表5 BaseAlgorithm 属性 描述 是否必填 数据类型 id 算法管理的算法ID 否 str subscription_id 订阅算法的订阅ID 否 str item_version_id 订阅算法的版本号 否 str code_dir 代码目录 否 str、Placeholder、Storage boot_file 启动文件 否 str、Placeholder、Storage command 启动命令 否 str、Placeholder parameters 算法超参 否 AlgorithmParameters的列表 engine 作业使用的镜像信息 否 JobEngine environments 环境变量 否 dict 表6 Algorithm 属性 描述 是否必填 数据类型 algorithm_id 算法管理的算法ID 是 str parameters 算法超参 否 Algorithm Parameters的列表 表7 AIGalleryAlgorithm 属性 描述 是否必填 数据类型 subscription_id 订阅算法的订阅ID 是 str item_version_id 订阅算法的版本号 是 str parameters 算法超参 否 Algorithm Parameters的列表 表8 AlgorithmParameters 属性 描述 是否必填 数据类型 name 算法超参的名称 是 str value 算法超参的值 是 int、bool、float、str、Placeholder、Storage 表9 JobEngine 属性 描述 是否必填 数据类型 engine_id 镜像ID 否 str、Placeholder engine_name 镜像名称 否 str、Placeholder engine_version 镜像版本 否 str、Placeholder image_url 镜像url 否 str、Placeholder 表10 JobSpec 属性 描述 是否必填 数据类型 resource 资源信息 是 JobResource log_export_path 日志输出路径 否 LogExportPath schedule_policy 作业调度配置策略 否 SchedulePolicy volumes 作业挂载的文件系统信息 否 list[Volume] 表11 JobResource 属性 描述 是否必填 数据类型 flavor 资源规格 是 Placeholder node_count 节点个数，默认为1，多节点表示支持分布式 否 int、Placeholder 表12 SchedulePolicy 属性 描述 是否必填 数据类型 priority 作业调度的优先级，仅支持配置为1、2、3，分别对应低、中、高三种优先级 是 int、Placeholder 表13 Volume 属性 描述 是否必填 数据类型 nfs NFS文件系统对象，在一个Volume对象中，nfs、pacific、pfs同时只能配置一个 否 NFS pacific pacific文件系统对象，在一个Volume对象中，nfs、pacific、pfs同时只能配置一个 否 Placeholder pfs OBS并行文件系统对象，在一个Volume对象中，nfs、pacific、pfs同时只能配置一个 否 PFS、Placeholder 表14 NFS 属性 描述 是否必填 数据类型 nfs_server_path NFS文件系统的服务地址 是 str、Placeholder local_path 挂载到容器里面的路径 是 str、Placeholder read_only 是否只读的方式挂载 否 bool、Placeholder 表15 PFS 属性 描述 是否必填 数据类型 pfs_path 并行文件系统的路径 是 str、Placeholder local_path 挂载到容器里面的路径 是 str、Placeholder

使用案例 场景一：基于数据集发布版本 使用场景：当数据集更新了数据时，可以通过该节点发布新的数据集版本供后续的节点使用。 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()

使用案例 主要包含三种场景的用例： 场景一：将指定存储路径下的数据导入到目标数据集中。适用于需要对数据集进行数据更新的操作。 用户将指定路径下已标注的数据导入到数据集中（同时导入标签信息），后续可增加数据集版本发布节点进行版本发布。 数据准备：提前在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] )

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

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] )

使用案例 主要包含两种场景的用例。 基于未标注数据创建数据集 基于已标注的数据创建数据集，并自动导入标注信息 基于未标注数据创建数据集 数据准备：存储在OBS文件夹中的未标注的数据。 from modelarts import workflow as wf # 通过CreateDatasetStep将存储在OBS中的数据创建成一个新版数据集 # 定义数据集输出路径参数 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="output_name", config=wf.data.OBSOutputConfig(obs_path=dataset_output_path)),# CreateDatasetStep的输出 properties=wf.steps.DatasetProperties( dataset_name=dataset_name, # 该名称对应的数据集如果不存在，则创建新的数据集;如果已存在，则直接使用该名称对应的数据集 data_type=wf.data.DataTypeEnum.IMAGE, # 数据集对应的数据类型, 示例为图像 ) ) # 注意dataset_name这个参数配置的数据集名称需要用户自行确认在该账号下未被他人使用，否则会导致期望的数据集未被创建，而后续节点错误使用了他人创建的数据集 workflow = wf.Workflow( name="create-dataset-demo", desc="this is a demo workflow", steps=[create_dataset] ) 基于已标注数据创建数据集，并导入标注信息 数据准备：存储在OBS文件夹中的已标注数据。 OBS目录导入已标注数据的规范：可参见OBS目录导入数据规范说明。 from modelarts import workflow as wf # 通过CreateDatasetStep将存储在OBS中的数据创建成一个新版数据集 # 定义数据集输出路径参数 dataset_output_path = wf.Placeholder(name="dataset_placeholder_name", placeholder_type=wf.PlaceholderType.STR, placeholder_format="obs") # 定义数据集名称参数 dataset_name = wf.Placeholder(name="dataset_placeholder_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="output_name", config=wf.data.OBSOutputConfig(obs_path=dataset_output_path)),# CreateDatasetStep的输出 properties=wf.steps.DatasetProperties( dataset_name=dataset_name, # 该名称对应的数据集如果不存在，则创建新的数据集;如果已存在，则直接使用该名称对应的数据集 data_type=wf.data.DataTypeEnum.IMAGE, # 数据集对应的数据类型, 示例为图像 import_config=wf.steps.ImportConfig( annotation_format_config=[ wf.steps.AnnotationFormatConfig( format_name=wf.steps.AnnotationFormatEnum.MA_IMAGE_CLASSIFICATION_V1, # 已标注数据的标注格式 scene=wf.data.LabelTaskTypeEnum.IMAGE_CLASSIFICATION) # 标注的场景类型 ] ) ) ) # 注意dataset_name这个参数配置的数据集名称需要用户自行确认在该账号下未被他人使用，否则会导致期望的数据集未被创建，而后续节点错误使用了他人创建的数据集 workflow = wf.Workflow( name="create-dataset-demo", desc="this is a demo workflow", steps=[create_dataset] )

属性总览 您可以使用CreateDatasetStep来构建数据集创建节点，CreateDatasetStep及相关对象结构如下。 表1 CreateDatasetStep 属性 描述 是否必填 数据类型 name 数据集创建节点的名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)，一个Workflow里的两个step名称不能重复。 是 str inputs 数据集创建节点的输入列表。 是 CreateDatasetInput或者CreateDatasetInput的列表 outputs 数据集创建节点的输出列表。 是 CreateDatasetOutput或者CreateDatasetOutput的列表 properties 数据集创建相关的配置信息。 是 DatasetProperties title title信息，主要用于前端的名称展示。 否 str description 数据集创建节点的描述信息。 否 str policy 节点执行的policy。 否 StepPolicy depend_steps 依赖的节点列表。 否 Step或者Step的列表 表2 CreateDatasetInput 属性 描述 是否必填 数据类型 name 数据集创建节点的输入名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输入名称不能重复。 是 str data 数据集创建节点的输入数据对象。 是 OBS相关对象，当前仅支持OBSPath、OBSConsumption、OBSPlaceholder、DataConsumptionSelector 表3 CreateDatasetOutput 属性 描述 是否必填 数据类型 name 数据集创建节点的输出名称，命名规范(只能包含英文字母、数字、下划线（_）、中划线（-），并且只能以英文字母开头，长度限制为64字符)。同一个Step的输出名称不能重复。 是 str config 数据集创建节点的输出相关配置。 是 当前仅支持OBSOutputConfig 表4 DatasetProperties 属性 描述 是否必填 数据类型 dataset_name 数据集的名称，只能是中文、字母、数字、下划线或中划线组成的合法字符串，长度为1-100位。 是 str、Placeholder dataset_format 数据集格式，默认为0，表示文件类型。 否 0：文件类型 1：表格类型 data_type 数据类型，默认为FREE_FORMAT。 否 DataTypeEnum description 描述信息。 否 str import_data 是否要导入数据，当前只支持表格数据，默认为False。 否 bool work_path_type 数据集输出路径类型，当前仅支持OBS，默认为0。 否 int import_config 标签导入的相关配置，默认为None，当基于已标注的数据创建数据集时，可指定该字段导入相关标注信息。 否 ImportConfig 表5 Importconfig 属性 描述 是否必填 数据类型 import_annotations 是否自动导入输入目录下的标注信息，支持检测/图像分类/文本分类。可选值如下： true：导入输入目录下的标注信息（默认值） false：不导入输入目录下的标注信息 否 str、Placeholder import_type 导入方式。可选值如下： dir：目录导入 manifest：按manifest文件导入 否 0：文件类型ImportTypeEnum annotation_format_config 导入的标注格式的配置参数。 否 DAnnotationFormaTypeEtConumfig的列表 表6 AnnotationFormatConfig 属性 描述 是否必填 数据类型 format_name 标注格式的名称。 否 AnnotationFormatEnum scene 标注场景，可选参数。 否 LabelTaskTypeEnum 枚举类型 枚举值 ImportTypeEnum DIR MANIFEST DataTypeEnum IMAGE TEXT AUDIO TABULAR VIDEO FREE_FORMAT AnnotationFormatEnum MA_IMAGE_CLASSIFICATION_V1 MA_IMAGENET_V1 MA_PASCAL_VOC_V1 YOLO MA_IMAGE_SEGMENTATION_V1 MA_TEXT_CLASSIFICATION_COMBINE_V1 MA_TEXT_CLASSIFICATION_V1 MA_AUDIO_CLASSIFICATION_DIR_V1