华为云用户手册

操作流程 通过VPN实现客户端远程接入VPC的操作流程如图1所示。 图1 操作流程 表2 操作流程说明 序号 步骤 说明 1 步骤1：创建VPN网关 VPN网关需要绑定EIP作为出口公网IP。 如果您已经购买EIP，则此处可以直接绑定使用。 2 步骤2：配置服务端 指定客户端需要访问的网段（本端网段）和客户端访问时使用的网段（客户端网段）。 选择服务端证书和客户端CA证书，用于建立VPN连接时的身份认证。 配置VPN连接的SSL参数（协议、端口、认证算法、加密算法等）。 3 步骤3：配置客户端 从控制台下载客户端配置，对配置文件进行修改后导入对应的VPN客户端软件。 4 步骤4：验证连通性 打开客户端设备的命令行窗口，执行ping命令，验证连通性。

操作步骤 在云日志服务管理控制台，单击“日志管理”。 在日志组列表中，单击接入ELB日志的日志组名称，进入日志流详情页面。 或者在日志流列表中，单击接入ELB日志的日志流名称，进入日志流详情页面。 在右上角选择时间范围。 在搜索框输入“status 500”，单击查询，开始搜索。 显示近一小时内ELB日志错误码为500的日志，status字段开启快速分析后会同步现在执行正常和异常状态占比。 单击查看原始日志的上下文，了解该问题发生时的应用运行状态。 设置日志结构化后，下拉框会同步显示系统内置字段及结构化配置的字段。 创建快速分析后，LTS会统计对应字段在日志流的日志文件中出现的条数，并生成日志指标（百分比）。

新建可视化图表 SQL查询与分析。 登录云日志服务管理控制台，在左侧导航栏中，选择“日志管理”。 在日志列表中，单击日志组名称前对应的，选择目标日志流，进入日志详情页面。 选择“可视化”，在SQL查询条件框中，选择对应时间并输入SQL语句，单击“查询”进行日志搜索。 当设置时间范围内日志量超过10亿行时会触发迭代查询，可以通过迭代查询分多次完成全部日志的查询，界面会显示“查询状态：结果精确”。 根据SQL查询返回的数据，依照您的业务需求选择不同图表类型，呈现查询结果。 关于更多SQL查询的说明，请参见SQL查询语法。 新建图表。 单击“新建”，新建为可视化图表。 或单击“保存”，将当前查询结果新建为可视化图表。当选中某个可视化图表时，单击“保存”，可对当前图表修改结果进行保存。 在创建图表页面中，配置相关参数。 如果开启“同时添加到仪表盘”按钮，新建图表可以直接添加到仪表盘中。 完成后，单击“确定”。 查看可视化图表。 单击“展开图表”，查看可视化图表。

将图表添加到仪表盘 将图表添加到仪表盘有两种方式： 方式一： 鼠标悬浮可视化图表名称，单击选择“添加到仪表盘”。 在弹出的移动图表页面中，选择待存放的仪表盘。 完成后，单击“确定”。 方式二： 创建仪表盘 登录云日志服务管理控制台，在左侧导航栏中，选择“仪表盘”，单击左上角下拉框。 单击“添加仪表盘”，在创建仪表盘页面，配置相关参数。 关于仪表盘参数的说明，请参见仪表盘。 将图表添加到仪表盘。 单击“开始添加图表”，进入添加可视化图表界面，选择目标日志新建的可视化图表。 完成后，单击“确定”。

SQL语法支持的数据类型 SQL查询中支持的数据类型如表1。如果当前字段数据类型需要改为其他数据类型，我们会进行数据类型的转换。例如STRING类型的字段转为LONG类型。字段数据类型转换之后的结果将会显示默认值，如STRING类型的数据转换为LONG类型的数据，结果会显示为LONG类型的默认值0。同理，当空值被转换为非空类型值时，也会使用默认值进行替换。例如，当把STRING类型空值转换为数字类型时，将会返回默认值0。 SQL语法中，字符必须被单引号（''）包裹，无符号或双引号（""）包裹的为字段或表名称，如：'msg'表示字符串msg，msg或"msg"表示日志结构化msg字段。 表1 SQL查询支持的数据类型 原生数据类型 默认值 说明 STRING "" 原生STRING类型 FLOAT 0.0 原生FLOAT类型 LONG 0 原生LONG类型 父主题： SQL查询语法

使用案例 控制单节点的执行 通过参数配置实现 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文件格式要求可参考分支控制部分，并且在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 } } ] 控制多分支的部分执行 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 ) 案例中job_step_a和job_step_b均配置了跳过策略，并且都使用参数进行控制。当参数值配置不同时，model_step的执行可以分为以下几种情况（model_step没有配置跳过策略，因此会遵循默认规则）： job_step_a_is_skip参数值 job_step_b_is_skip参数值 model_step是否执行 True True 跳过 False 执行 False True 执行 False 执行 默认规则：当某个节点依赖的所有节点状态均为跳过时，该节点自动跳过，否则正常执行，此判断逻辑可扩展至任意节点。 在上述案例的基础上，如果需要打破默认规则，在job_step_a以及job_step_b跳过时，model_step也允许执行，则只需要在model_step中也配置跳过策略即可（跳过策略的优先级高于默认规则）。

异步推理服务部署相关信息配置 在ModelArts管理控制台，左侧菜单栏选择“Workflow”进入Workflow页面。 在服务部署节点启动之后会等待用户设置相关配置信息，选择AI应用及版本为异步推理模型，设置服务启动参数，配置完成后直接单击继续运行即可。 图1 选择异步推理AI应用 其中服务启动参数与您选择的异步推理AI应用相关，选择了需要的AI应用及版本后，系统会自动匹配响应的服务启动参数。

MLOps功能介绍 机器学习开发流程主要可以定义为四个步骤：项目设计、数据工程、模型构建、部署落地。AI开发并不是一个单向的流水线作业，在开发的过程中，会根据数据和模型结果进行多轮的实验迭代。算法工程师会根据数据特征以及数据的标签做多样化的数据处理以及多种模型优化，以获得在已有的数据集上更好的模型效果。传统的AI应用交付会直接在实验迭代结束后以输出的模型为终点。当应用上线后，随着时间的推移，会出现模型漂移的问题。新的数据和新的特征在已有的模型上表现会越来越差。在MLOps中，实验迭代的产物将会是一条固化下来的流水线，这条流水线将会包含数据工程、模型算法、训练配置等。用户将会使用这条流水线在持续产生的数据中持续迭代训练，确保这条流水线生产出来的模型的AI应用始终维持在一个较好的状态。 图1 MLOps MLOps的整条链路需要有一个工具去承载，MLOps打通了算法开发到交付运维的全流程。和以往的开发交付不同，以往的开发与交付过程是分离的，算法工程师开发完的模型，一般都需要交付给下游系统工程师。MLOps和以往的开发交付不同，在这个过程中，算法工程师参与度还是非常高的。企业内部一般都是有一个交付配合的机制。从项目管理角度上需要增加一个AI项目的工作流程机制管理，流程管理不是一个简单的流水线构建管理，它是一个任务管理体系。 这个工具需要具备以下的能力： 流程分析：沉淀行业样例流水线，帮助用户能快速进行AI项目的参考设计，启动快速的AI项目流程设计。 流程定义与重定义：以流水线作为承载项，用户能快速定义AI项目，实现训练+推理上线的工作流设计。 资源分配：支持账号管理机制给流水线中的参与人员（包含开发者和运维人员）分配相应的资源配额与权限，并查看相应的资源使用情况等。 时间安排：围绕子流水线配置相应的子任务安排，并加以通知机制，实现流程执行过程之间配合的运转高效管理。 流程质量与效率测评：提供流水线的任务执行过程视图，增加不同的检查点，如数据评估、模型评估、性能评估等，让AI项目管理者能很方便的查看流水线执行过程的质量与效率。 流程优化：围绕流水线每一次迭代，用户可以自定义输出相关的核心指标，并获取相应的问题数据与原因等，从而基于这些指标，快速决定下一轮迭代的执行优化。

什么是MLOps MLOps(Machine Learning Operation)是“机器学习”（Machine Learning）和“DevOps”（Development and Operations）的组合实践。随着机器学习的发展，人们对它的期待不仅仅是学术研究方面的领先突破，更希望这些技术能够系统化地落地到各个场景中。但技术的真实落地和学术研究还是有比较大的差别的。在学术研究中，一个AI算法的开发是面向固定的数据集（公共数据集或者某个特定场景固定数据集），基于单个数据集，不断做算法的迭代与优化。面向场景的AI系统化开发的过程中，除了模型的开发，还有整套系统的开发，于是软件系统开发中成功经验“DevOps”被自然地引入进来。但是，在人工智能时代，传统的DevOps已经不能完全覆盖一个人工智能系统开发的全流程了。

DevOps DevOps，即Development and Operations，是一组过程、方法与系统的统称，用于促进软件开发、运维和质量保障部门之间的沟通、协作与整合。在大型的软件系统开发中，DevOps被验证是一个非常成功的方法。DevOps不仅可以加快业务与开发之间的互动与迭代，还可以解决开发与运维之间的冲突。开发侧很快，运维侧太稳，这个就是常说的开发与运维之间固有的、根因的冲突。在AI应用落地的过程中，也有类似的冲突。AI应用的开发门槛较高，需要有一定的算法基础，而且算法需要快速高效地迭代。专业的运维人员追求的更多是稳定、安全和可靠；专业知识也和AI算法大相径庭。运维人员需要去理解算法人员的设计与思路才能保障服务，这对于运维人员来说，门槛更高了。在这种情况下，更多时候可能需要一个算法人员去端到端负责，这样一来，人力成本就会过高。这种模式在少量模型应用的场景是可行的，但是当规模化落地AI应用时，人力问题将会成为瓶颈。

使用订阅自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] )

使用模板的方式注册模型 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] )

从训练作业中注册模型（模型输入来源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来定义一个模型注册节点，输入来源于自定义镜像地址 # 定义镜像数据 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] )

发布到AI Gallery Workflow支持发布到gallery，分享给其他用户使用，执行如下代码即可完成发布。 workflow.release_to_gallery() 发布完成后可前往gallery查看相应的资产信息，资产权限默认为private，可在资产的console页面自行修改。 其中release_to_gallery()方法包含以下入参： 参数名称 描述 是否必填 参数类型 content_id Workflow资产ID 否 str version Workflow资产的版本号，格式为x.x.x 否 str desc Workflow资产版本的描述信息 否 str title Workflow资产名称，该参数未填写时默认使用Workflow的名称作为资产名称 否 str visibility Workflow资产可见性，支持"public"-公开、"group"-白名单、"private"-私有，仅自己可见三种，默认为"private"。 否 str group_users 白名单列表，仅支持填写domain_id，当visibility为"group"时才需要填写该字段 否 list[str] 根据方法的入参不同，主要可分为以下两种使用场景： Workflow.release_to_gallery(title="资产名称")发布Workflow新资产，版本号为"1.0.0"；如果Workflow包含非gallery的算法，则自动将依赖算法发布至gallery，版本号为"1.0.0"。 Workflow.release_to_gallery(content_id="**", title="资产名称")基于指定的Workflow资产，发布新的版本，版本号自动增加；如果Workflow包含gallery的算法，则自动将依赖的算法资产发布新版本，版本号也自动增加。 Workflow资产白名单设置： 在资产第一次发布时，可以通过release_to_gallery方法的visibility+group_users字段进行设置，后续需要对指定资产进行用户白名单添加或删除操作时，可执行如下命令： from modelarts import workflow as wf # 添加指定的白名单用户列表 wf.add_whitelist_users(content_id="**", version_num="*.*.*", user_groups=["**", "**"]) # 删除指定的白名单用户列表 wf.delete_whitelist_users(content_id="**", version_num="*.*.*", user_groups=["**", "**"]) 在给Workflow资产添加或删除指定白名单用户列表时，会自动查询该版本依赖的算法资产信息，同步对算法资产进行相应的白名单设置。 父主题： 发布Workflow

新增在线服务 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] )

使用案例 统一存储主要用于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对象需要在这里添加 ) 父主题： 统一存储

属性总览 您可以使用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 父主题： 数据集创建节点

运行态 Workflow提供了整套可视化的工作流运行方式，简称为运行态。使用者不需要了解工作流的内部细节，只需要关注一些简单的参数配置即可启动运行工作流。运行态的工作流来源主要为：通过开发态发布或者从AI Gallery订阅。 运行态主要提供以下能力： 统一配置管理：管理工作流需要配置的参数及使用的资源等。 操作工作流：启动、停止、重试、复制、删除工作流等。 运行记录：工作流历史运行的参数以及状态记录。 父主题： 核心概念

进阶示例 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文件格式要求可参考作业类型节点部分，并且在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 } } ]

资源规格查询 您在创建作业类型节点之前可以通过以下操作来获取该账号所支持的训练资源规格列表以及引擎规格列表： 导包 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) 父主题： 作业类型节点

属性总览 您可以使用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() 父主题： 数据集版本发布节点

属性总览 您可以使用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() # 主要在服务部署节点的输出中使用 如果您没有特殊需求，可直接使用内置的默认值。 父主题： 服务部署节点

编写工作流 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] ) 在工作流编写完成后可自行进行调试、发布等操作。