华为云用户手册

前提条件 为了确保负载均衡器与后端服务器进行正常通信和健康检查正常，添加后端服务器后必须检查后端服务器所在的安全组规则和网络ACL规则。 流量经共享型ELB转到后端服务器以后，源IP会被转换为100.125.0.0/16的IP。 后端服务器的安全组规则必须配置放行100.125.0.0/16网段。 网络ACL规则为子网级别的可选安全层，若ELB的后端子网关联了网络ACL，网络ACL规则必须配置允许源地址为ELB后端子网所属网段。 若共享型ELB实例开启“获取客户端IP”功能，共享型ELB四层监听器转发的流量将不受安全组规则和网络ACL限制，安全组规则和网络ACL规则均无需额外放通。建议您使用监听器的访问控制功能对访问IP进行限制。

验证负载均衡服务 负载均衡器配置完成后，可通过访问ELB实例对应的 域名 以及指定的URL，验证是否实现访问到不同的后端服务器。 修改本地PC的“C:\Windows\System32\drivers\etc\hosts”文件，将域名映射到创建的ELB实例的EIP上。 ELB实例的EIP请在负载均衡器的基本信息界面查看。 图10 本地PC的hosts文件 在本地PC的命令行窗口，执行以下命令测试域名映射ELB实例的EIP是否成功。 ping www.example.com 如有回复数据包，则说明域名映射成功。 使用浏览器访问“http://www.example.com/ELB01/”，显示如下页面，说明本次访问请求被ELB实例转发到弹性云服务器"E CS 01"，"ECS01"正常处理请求并返回请求的页面。 图11 访问到ECS01 "ELB01/"表示访问的是名为“ELB01”的默认目录，"ELB01"表示访问的是名为“ELB01”的文件。所以，此处"ELB01"后面的“/”符号必须保留。 使用浏览器访问“http://www.example.com/ELB02/”，显示如下页面，说明本次访问请求被ELB实例转发到弹性云服务器"ECS02"，"ECS02"正常处理请求并返回请求的页面。 图12 访问到ECS02

操作场景 假如您有两个Web应用，需要部署在两个弹性云服务器（简称ECS）上。对外提供同一个访问域名“www.example.com”，根据不同的URL（/ELB01/和/ELB02/）来决定需要访问的应用。 负载均衡可以配置基于域名和URL的转发策略，将来自不同域名或者不同URL的请求转发到不同的后端服务器组处理。此功能目前仅支持协议类型为HTTP、HTTPS的监听器。 本文以HTTP转发为例，通过配置ELB的转发策略，实现基于访问同一个域名下的不同的URL，ELB实例根据URL将访问请求转发到不同的后端服务器组进行处理。

创建 IAM 用户 如果您需要多用户协同操作管理您账号下的资源，为了避免共享您的密码/访问密钥，您可以通过IAM创建用户，并授予用户对应权限。这些用户可以使用特别的登录链接和自己单独的用户账号访问华为云平台，帮助您高效的管理资源，您还可以设置账号安全策略确保这些账号的安全，从而降低您的企业信息安全风险。 如果您已注册华为云平台账号但尚未为自己创建一个IAM用户，则可以使用IAM控制台自行创建。 以创建ELB管理员为例，具体步骤如下： 使用账号和密码登录管理控制台。 单击右上方登录的用户名，在下拉列表中选择“ 统一身份认证 ”。 在左侧导航栏中单击“用户”，在“用户”界面，单击“创建用户”。 在“创建用户”界面填写“用户信息”。 用户名：设置一个用户名，如“elb_administrator”。 邮箱：IAM用户绑定的邮箱，仅“访问方式”选择“首次登录时设置”时必填，选择其他访问方式时选填。 手机号（选填）：IAM用户绑定的手机号。 描述（选填）：输入用户描述信息，如“ELB管理员”。 在“创建用户”界面选择“访问方式”为“华为云管理控制台访问”，设置控制台登录密码为“自定义”并输入密码，完成后单击“下一步”。 图1 选择访问方式 ELB管理员用于登录管理控制台管理用户。如果您为自己创建ELB管理员，建议使用自定义方式设置密码。如果您为他人创建ELB管理员，建议使用“首次登录时设置”的方式，由用户自己设置密码。 （可选）将用户加入到“admin”用户组，完成后单击“下一步”。 “admin”用户组拥有所有操作权限，如果您想为IAM用户精细授权，请参阅创建用户并授权使用ELB。 创建成功后，用户列表中显示新创建的IAM用户。IAM用户可以使用列表上方的IAM用户登录链接登录控制台。

配置Redis目的端参数 表1 Redis作为目的端时的作业参数 参数类型 参数名 说明 取值样例 基本参数 redis键前缀 对应关系数据库的表名。 TABLENAME 值存储类型 存储类型分STRING、hash、list、set和zset。 STRING 写入前将相同的键删除 写入前是否将相同的键删除。 否 高级属性 键分隔符 用来分隔关系数据库的表和列名。 _ 值分隔符 以STRING方式存储，列之间的分隔符。 存储类型为列表时字串分割成数组的字符。 ; key值有效期 设置统一的生存时间。单位：秒。 3600 父主题： 配置作业目的端参数

配置HDFS目的端参数 表1 HDFS作为目的端时的作业参数 参数类型 参数名 说明 取值样例 基本参数 写入目录 写入数据到HDFS服务器的目录。 /user/cdm/output 文件格式 传输数据时使用的格式。其中CSV和JSON仅支持迁移到数据表场景，二进制格式适用于文件迁移场景。 CSV格式 换行符处理方式 指定在写入文本文件表的数据包含换行符，特指(\n|\r|\r\n)的情况下处理换行符的策略。 删除 不处理 替换为其他字符串 删除 换行符替换字符串 当换行符处理方式选择为替换时，指定替换的字符串。 - 高级属性 写入到临时文件 文件格式为二进制格式时显示该参数。 将二进制文件先写入到临时文件。临时文件以".tmp"作为后缀。 否 换行符 文件格式为CSV格式时显示该参数。 文件中的换行符，默认自动识别"\n"、"\r"或"\r\n"。手动配置特殊字符，如空格回车需使用URL编码后的值。或通过编辑作业json方式配置，无需URL编码。 \n 字段分隔符 文件格式为CSV格式时显示该参数。 文件中的字段分隔符。配置特殊字符需先url编码。 , 作业成功标识文件 标识文件名。 当作业成功时，在写入目录下生成标识文件。不输入文件名时不启用该功能。 finish.txt 使用包围符 文件格式为CSV格式时显示该参数。 使用包围符来括住字符串值。包围符内的字段分隔符被视为字符串值的一部分，目前只支持"作为包围符。： 否 自定义目录层次 支持用户自定义文件的目录层次。 例如：【表名】/【年】/【月】/【日】/【数据文件名】. csv 否 目录层次 自定义目录层次选择是时显示该参数。 指定文件的目录层次，支持时间宏（时间格式为yyyy/MM/dd）。源端为关系型数据库数据源时，目录层次为源端表名+自定义目录，其他场景下为自定义目录。 ${dateformat(yyyy/MM/dd, -1, DAY)} 文件名前缀 文件格式为CSV格式时显示该参数。 设置文件名前缀。 文件名格式：prefix-jobname-timestamp-index。 data 压缩格式 文件格式为CSV格式时显示该参数。 选择写入文件的压缩格式。 NONE DEFLATE GZIP BZIP2 SNAPPY SNAPPY 加密方式 文件格式为二进制格式时显示该参数。 对上传的数据进行加密。 无 AES-256-GCM 无 数据加密 密钥 文件格式为二进制格式且选择加密方式时显示该参数。 数据加密密钥（Data Encryption Key），AES-256-GCM密钥由长度64的十六进制数组成。 DD0AE00DFECD78BF051BCFDA25BD4E320DB0A7AC75A1F3FC3D3C56A457DCDC1B 初始化向量 文件格式为二进制格式且选择加密方式时显示该参数。 设置初始化向量，由长度32的十六进制数组成。 5C91687BA886EDCD12ACBC3FF19A3C3F 父主题： 配置作业目的端参数

操作步骤 登录 CDM 管理控制台。单击左侧导航上的“集群管理”，进入集群管理界面。 图1 集群列表 “创建来源”列仅通过 DataArts Studio 服务进入数据集成界面可以看到。 单击集群名称后，选择“标签”页签。 图2 修改集群配置 单击“添加/编辑标签”，通过添加、修改标签为CDM集群设置资源标识。 图3 添加标签 一个集群最多可添加10个标签。 标签键（key）的最大长度为36个字符，标签值（value）的最大长度为43个字符。 （可选）在标签列表中，单击标签操作列“删除”，删除CDM集群标签。 通过以下两种方式筛选出所配置标签的资源。 在标签管理服务中，选择资源搜索条件，单击“搜索”即可筛选出所配置标签的资源。 在集群列表中，单击标签搜索，筛选出所配置标签的资源。

指定文件名迁移 从FTP/SFTP/OBS导出文件时，CDM支持指定文件名迁移，用户可以单次迁移多个指定的文件（最多50个），导出的多个文件只能写到目的端的同一个目录。 在创建表/文件迁移作业时，如果源端数据源为FTP/SFTP/OBS，CDM源端的作业参数“源目录或文件”支持输入多个文件名（最多50个），文件名之间默认使用“|”分隔，您也可以自定义文件分隔符，从而实现文件列表迁移。 迁移文件或对象时支持文件级增量迁移（通过配置跳过重复文件实现），但不支持断点续传。 例如要迁移3个文件，第2个文件迁移到一半时由于网络原因失败，再次启动迁移任务时，会跳过第1个文件，从第2个文件开始重新传，但不能从第2个文件失败的位置重新传。 文件迁移时，单个任务支持千万数量的文件，如果待迁移目录下文件过多，建议拆分到不同目录并创建多个任务。 父主题： 关键操作指导

Django日志 日志样例： [08/Jan/2018 20:59:07 ] settings INFO Welcome to Hue 3.9.0 正则表达式为： ^\[(.*)\] (\w*) (\w*) (.*).* 解析结果如下： 表4 Django日志解析结果 列号 样值 1 08/Jan/2018 20:59:07 2 settings 3 INFO 4 Welcome to Hue 3.9.0

Apache server日志 日志样例： [Mon Jan 08 20:43:51.854334 2018] [mpm_event:notice] [pid 36465:tid 140557517657856] AH00489: Apache/2.4.12 (Unix) OpenSSL/1.0.1t configured -- resuming normal operations 正则表达式为： ^\[(.*)\] \[(.*)\] \[(.*)\] (.*).* 解析结果如下： 表5 Apache server日志解析结果 列号 样值 1 Mon Jan 08 20:43:51.854334 2018 2 mpm_event:notice 3 pid 36465:tid 140557517657856 4 AH00489: Apache/2.4.12 (Unix) OpenSSL/1.0.1t configured -- resuming normal operations

Log4J审计日志 日志样例： 2018-01-11 08:51:06,156 INFO [org.apache.sqoop.audit.FileAuditLogger.logAuditEvent(FileAuditLogger.java:61)] user=sqoop.anonymous.user ip=189.xxx.xxx.75 op=show obj=version objId=x 正则表达式为： ^(\d.*\d) (\w*) \[(.*)\] user=(\w.*) ip=(\w.*) op=(\w.*) obj=(\w.*) objId=(.*).* 解析结果如下： 表2 Log4J审计日志解析结果 列号 样值 1 2018-01-11 08:51:06,156 2 INFO 3 org.apache.sqoop.audit.FileAuditLogger.logAuditEvent(FileAuditLogger.java:61) 4 sqoop.anonymous.user 5 189.xxx.xxx.75 6 show 7 version 8 x

Tomcat日志 日志样例： 11-Jan-2018 09:00:06.907 INFO [main] org.apache.catalina.startup.VersionLoggerListener.log OS Name: Linux 正则表达式为： ^(\d.*\d) (\w*) \[(.*)\] ([\w\.]*) (\w.*).* 解析结果如下： 表3 Tomcat日志解析结果 列号 样值 1 11-Jan-2018 09:00:06.907 2 INFO 3 main 4 org.apache.catalina.startup.VersionLoggerListener.log 5 OS Name:Linux

Log4J日志 日志样例： 2018-01-11 08:50:59,001 INFO [org.apache.sqoop.core.SqoopConfiguration.configureClassLoader(SqoopConfiguration.java:251)] Adding jars to current classloader from property: org.apache.sqoop.classpath.extra 正则表达式为： ^(\d.*\d) (\w*) \[(.*)\] (\w.*).* 解析出的结果如下： 表1 Log4J日志解析结果 列号 样值 1 2018-01-11 08:50:59,001 2 INFO 3 org.apache.sqoop.core.SqoopConfiguration.configureClassLoader(SqoopConfiguration.java:251) 4 Adding jars to current classloader from property: org.apache.sqoop.classpath.extra

KMS加密 源端解密不支持KMS。 CDM目前只支持导入文件到OBS时，目的端使用KMS加密，表/文件迁移和整库迁移都支持。在“目的端作业配置”的“高级属性”中配置。 KMS密钥需要先在数据加密服务创建，具体操作请参见《数据加密服务 用户指南》。 当启用KMS加密功能后，用户上传对象时，数据会加密成密文存储在OBS。用户从OBS下载加密对象时，存储的密文会先在OBS服务端解密为明文，再提供给用户。 如果选择使用KMS加密，则无法使用MD5校验一致性。 如果这里使用其它项目的KMS ID，则需要修改“项目ID”参数为KMS ID所属的项目ID；如果KMS ID与CDM在同一个项目下，“项目ID”参数保持默认即可。 使用KMS加密后，OBS上对象的加密状态不可以修改。 使用中的KMS密钥不可以删除，如果删除将导致加密对象不能下载。

AES-256-GCM加密 目前只支持AES-256-GCM（NoPadding）。该加密算法在目的端为加密，在源端为解密，支持的源端与目的端数据源如下。 源端支持的数据源：HDFS（使用二进制格式传输时支持）。 目的端支持的数据源：HDFS（使用二进制格式传输时支持）。 下面分别以HDFS导出加密文件时解密、导入文件到HDFS时加密为例，介绍AES-256-GCM加解密的使用方法。 源端配置解密 创建从HDFS导出文件的CDM作业时，源端数据源选择HDFS、文件格式选择二进制格式后，在“源端作业配置”的“高级属性”中，配置如下参数。 加密方式：选择“AES-256-GCM”。 数据加密密钥：这里的密钥必须与加密时配置的密钥一致，否则解密出来的数据会错误，且系统不会提示异常。 初始化向量：这里的初始化向量必须与加密时配置的初始化向量一致，否则解密出来的数据会错误，且系统不会提示异常。 这样CDM从HDFS导出加密过的文件时，写入目的端的文件便是解密后的明文文件。 目的端配置加密 创建CDM导入文件到HDFS的作业时，目的端数据源选择HDFS、文件格式选择二进制格式后，在“目的端作业配置”的“高级属性”中，配置如下参数。 加密方式：选择“AES-256-GCM”。 数据加密密钥：用户自定义密钥，密钥由长度64的十六进制数组成，不区分大小写但必须64位，例如“DD0AE00DFECD78BF051BCFDA25BD4E320DB0A7AC75A1F3FC3D3C56A457DCDC1B”。 初始化向量：用户自定义初始化向量，初始化向量由长度32的十六进制数组成，不区分大小写但必须32位，例如“5C91687BA886EDCD12ACBC3FF19A3C3F”。 这样在CDM导入文件到HDFS时，目的端HDFS上的文件便是经过AES-256-GCM算法加密后的文件。

事务模式迁移 CDM的事务模式迁移，是指当CDM作业执行失败时，将数据回滚到作业开始之前的状态，自动清理目的表中的数据。 参数位置：创建表/文件迁移的作业时，如果目的端为关系型数据库，在目的端作业配置的高级属性中，可以通过“先导入阶段表”参数选择是否启用事务模式。 参数原理：如果启用，在作业执行时CDM会自动创建临时表，先将数据导入到该临时表，导入成功后再通过数据库的事务模式将数据迁移到目标表中；导入失败则将目的表回滚到作业开始之前的状态。 图1 事务模式迁移 如果“导入开始前”选择“清除部分数据”或“清除全部数据”，CDM的事务模式不会回滚已经删除的数据。 父主题： 关键操作指导

操作场景 CDM集群创建完成后，支持解绑或绑定EIP。EIP即弹性公网IP，由虚拟私有云（Virtual Private Cloud，简称VPC）负责其计费。 如果CDM需要访问本地数据源、Internet的数据源，或者跨VPC的云服务，则必须要为CDM集群绑定一个弹性IP，或者使用NAT网关让CDM集群与其他弹性云服务器共享弹性IP访问Internet，具体操作请见添加SNAT规则。 如果用户对本地数据源的访问通道做了SSL加密，则CDM无法通过弹性IP连接数据源。

MD5校验文件一致性 CDM数据迁移以抽取-写入模式进行，CDM首先从源端抽取数据，然后将数据写入到目的端。在迁移文件到OBS时，迁移模式如图1所示。 图1 迁移文件到OBS 在这个过程中，CDM支持使用MD5检验文件一致性。 抽取时 该功能支持源端为OBS、HDFS、FTP、SFTP、HTTP。可校验CDM抽取的文件，是否与源文件一致。 该功能由源端作业参数“MD5文件名后缀”控制（“文件格式”为“二进制格式”时生效），配置为源端文件系统中的MD5文件名后缀。 当源端数据文件同一目录下有对应后缀的保存md5值的文件，例如build.sh和build.sh.md5在同一目录下。若配置了“MD5文件名后缀”，则只迁移有MD5值的文件至目的端，没有MD5值或者MD5不匹配的数据文件将迁移失败，MD5文件自身不被迁移。 若未配置“MD5文件名后缀”，则迁移所有文件。 写入时 该功能目前只支持目的端为OBS。可校验写入OBS的文件，是否与CDM抽取的文件一致。 该功能由目的端作业参数“校验MD5值”控制，读取文件后写入OBS时，通过HTTP Header将MD5值提供给OBS做写入校验，并将校验结果写入OBS桶（该桶可以不是存储迁移文件的桶）。如果源端没有MD5文件则不校验。 迁移文件到文件系统时，目前只支持校验CDM抽取的文件是否与源文件一致（即只校验抽取的数据）。 迁移文件到OBS时，支持抽取和写入文件时都校验。 如果选择使用MD5校验，则无法使用KMS加密。 父主题： 关键操作指导

查看集群基本信息 登录CDM管理控制台。单击左侧导航上的“集群管理”，进入集群管理界面。 或参考访问DataArts Studio实例控制台，登录DataArts Studio管理控制台。在DataArts Studio控制台首页，选择对应工作空间的“数据集成”模块，进入CDM首页。 图1 集群列表 “创建来源”列仅通过DataArts Studio服务进入数据集成界面可以看到。 单击集群名称，可查看集群的基本信息。 图2 CDM集群的配置信息

操作场景 CDM集群已经创建成功后，您可以查看集群基本信息，并修改集群的配置。 查看集群基本信息： 集群信息：集群版本、创建时间、项目ID、实例ID和集群ID等。 节点配置：集群规格、CPU和内存配置等信息。 网络信息：网络配置。 支持修改集群的以下配置： 消息通知 ：CDM的迁移作业（目前仅支持表/文件迁移的作业）失败时，或者EIP异常时，会发送短信或邮件通知用户。该功能产生的消息通知不会计入收费项。 用户隔离：控制其他用户是否能够查看、操作该集群中的迁移作业和连接。 开启该功能时，该集群中的迁移作业、连接会被隔离，华为账号下的其他IAM用户无法查看、操作该集群中的迁移作业和连接。 按组批量启动作业会运行组内所有作业。如果开启了用户隔离功能，即使华为账号下的其他IAM用户无法查看到组内作业，按组批量启动作业依然会将组内作业运行，因此在用户隔离场景不建议使用按组批量启动作业功能。 关闭该功能时，该集群中的迁移作业、连接信息可以用户共享，华为账号下的所有拥有相应权限的IAM用户可以查看、操作迁移作业和连接。 注意，用户隔离关闭后需要重启集群VM才能生效。 最大抽取并发数：限制作业运行的总抽取并发数，如果当前所有作业总并发数超出限制，超出部分将排队等待。 注意，最大抽取并发数取值范围为1-1000，建议根据集群规格进行配置，建议值详见最大抽取并发数。过高的并发数可能导致内存溢出，请谨慎修改。 此处的“最大抽取并发数”参数与作业配置管理处的“最大抽取并发数”参数同步，在任意一处修改即可生效。

检测规则 卡死检测主要是通过监控作业进程的状态和资源利用率来判定作业是否卡死。会启动一个进程来周期性地监控上述两个指标的变化情况。 进程状态：只要训练作业中存在进程IO有变化，进入下一个检测周期。如果在多个检测周期内，作业所有进程IO都没有变化，则进入资源利用率检测阶段。 资源利用率：在作业进程IO没有变化的情况下，采集一定时间段内的GPU利用率，并根据这段时间内的GPU利用率的方差和中位数来判断资源使用率是否有变化。如果没有变化，则判定作业卡死。

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

触发容错环境检测达到的效果 容错检查正常通过时，会打印检测项目的日志，表示具体涉及的检查项目成功。您可以通过在日志中搜索“item”关键字查看。当容错检查正常通过时，可以减少运行故障上报问题。 容错检查失败时，会打印检查失败的日志。您可以通过在日志中搜索“item”关键字查看失败信息。 如果作业重启次数没有达到设定的次数，则会自动做重新下发作业。您可以通过搜索“error,exiting”关键字查找作业重启失败结束的日志。

开启容错检查 用户可以在创建训练作业时通过设置自动重启的方式开启容错检查。 使用ModelArts控制台的创建训练作业页面设置自动重启： 用户可以在控制台页面通过开关的方式开启自动重启。“自动重启”开关默认不开启，表示不做重新下发作业，也不会启用环境检测。打开开关后，允许设置重启次数为1~128次。 图5 自动重启设置 使用API接口设置容错检查： 用户可以通过API接口的方式开启自动重启。创建训练作业时，在“metadata”字段的“annotations”中传入“fault-tolerance/job-retry-num”字段。 添加“fault-tolerance/job-retry-num”字段，视为开启自动重启，value的范围可以设置为1~128的整数。value值表示最大允许重新下发作业的次数。如果不传入则默认为0，表示不做重新下发作业，也不会启用环境检测。 图6 设置API

创建训练作业 本案例创建训练作业时，需要配置如下参数。 表1 创建训练作业的配置说明 参数名称 说明 “创建方式” 选择“自定义算法”。 “启动方式” 选择“自定义”。 “镜像” 选择用于训练的 自定义镜像 。 “代码目录” 执行本次训练作业所需的代码目录。本文示例的代码目录为“obs://test-modelarts/ascend/code/”。 “启动命令” 镜像的Python启动命令。本文示例的启动命令为“bash ${MA_JOB_DIR}/code/run_torch_ddp_npu.sh”。其中，启动脚本的完整代码请参见代码示例。

代码示例 文件目录结构如下所示，将以下文件上传至OBS桶中： code # 代码根目录 └─torch_ddp.py # PyTorch DDP训练代码文件 └─main.py # 使用PyTorch预置框架功能，通过mp.spawn命令启动训练的启动文件 └─torchlaunch.sh # 使用自定义镜像功能，通过torch.distributed.launch命令启动训练的启动文件 └─torchrun.sh # 使用自定义镜像功能，通过torch.distributed.run命令启动训练的启动文件 torch_ddp.py内容如下： import os import torch import torch.distributed as dist import torch.nn as nn import torch.optim as optim from torch.nn.parallel import DistributedDataParallel as DDP # 用于通过 mp.spawn 启动 def init_from_arg(local_rank, base_rank, world_size, init_method): rank = base_rank + local_rank dist.init_process_group("nccl", rank=rank, init_method=init_method, world_size=world_size) ddp_train(local_rank) # 用于通过 torch.distributed.launch 或 torch.distributed.run 启动 def init_from_env(): dist.init_process_group(backend='nccl', init_method='env://') local_rank=int(os.environ["LOCAL_RANK"]) ddp_train(local_rank) def cleanup(): dist.destroy_process_group() class ToyModel(nn.Module): def __init__(self): super(ToyModel, self).__init__() self.net1 = nn.Linear(10, 10) self.relu = nn.ReLU() self.net2 = nn.Linear(10, 5) def forward(self, x): return self.net2(self.relu(self.net1(x))) def ddp_train(device_id): # create model and move it to GPU with id rank model = ToyModel().to(device_id) ddp_model = DDP(model, device_ids=[device_id]) loss_fn = nn.MSELoss() optimizer = optim.SGD(ddp_model.parameters(), lr=0.001) optimizer.zero_grad() outputs = ddp_model(torch.randn(20, 10)) labels = torch.randn(20, 5).to(device_id) loss_fn(outputs, labels).backward() optimizer.step() cleanup() if __name__ == "__main__": init_from_env() main.py内容如下： import argparse import torch import torch.multiprocessing as mp parser = argparse.ArgumentParser(description='ddp demo args') parser.add_argument('--world_size', type=int, required=True) parser.add_argument('--rank', type=int, required=True) parser.add_argument('--init_method', type=str, required=True) args, unknown = parser.parse_known_args() if __name__ == "__main__": n_gpus = torch.cuda.device_count() world_size = n_gpus * args.world_size base_rank = n_gpus * args.rank # 调用 DDP 示例代码中的启动函数 from torch_ddp import init_from_arg mp.spawn(init_from_arg, args=(base_rank, world_size, args.init_method), nprocs=n_gpus, join=True) torchlaunch.sh内容如下： #!/bin/bash # 系统默认环境变量，不建议修改 MASTER_HOST="$VC_WORKER_HOSTS" MASTER_ADDR="${VC_WORKER_HOSTS%%,*}" MASTER_PORT="6060" JOB_ID="1234" NNODES="$MA_NUM_HOSTS" NODE_RANK="$VC_TASK_INDEX" NGPUS_PER_NODE="$MA_NUM_GPUS" # 自定义环境变量，指定python脚本和参数 PYTHON_SCRIPT=${MA_JOB_DIR}/code/torch_ddp.py PYTHON_ARGS="" CMD="python -m torch.distributed.launch \ --nnodes=$NNODES \ --node_rank=$NODE_RANK \ --nproc_per_node=$NGPUS_PER_NODE \ --master_addr $MASTER_ADDR \ --master_port=$MASTER_PORT \ --use_env \ $PYTHON_SCRIPT \ $PYTHON_ARGS " echo $CMD $CMD torchrun.sh内容如下： PyTorch 2.1版本需要将“rdzv_backend”参数设置为“static：--rdzv_backend=static”。 #!/bin/bash # 系统默认环境变量，不建议修改 MASTER_HOST="$VC_WORKER_HOSTS" MASTER_ADDR="${VC_WORKER_HOSTS%%,*}" MASTER_PORT="6060" JOB_ID="1234" NNODES="$MA_NUM_HOSTS" NODE_RANK="$VC_TASK_INDEX" NGPUS_PER_NODE="$MA_NUM_GPUS" # 自定义环境变量，指定python脚本和参数 PYTHON_SCRIPT=${MA_JOB_DIR}/code/torch_ddp.py PYTHON_ARGS="" if [[ $NODE_RANK == 0 ]]; then EXT_ARGS="--rdzv_conf=is_host=1" else EXT_ARGS="" fi CMD="python -m torch.distributed.run \ --nnodes=$NNODES \ --node_rank=$NODE_RANK \ $EXT_ARGS \ --nproc_per_node=$NGPUS_PER_NODE \ --rdzv_id=$JOB_ID \ --rdzv_backend=c10d \ --rdzv_endpoint=$MASTER_ADDR:$MASTER_PORT \ $PYTHON_SCRIPT \ $PYTHON_ARGS " echo $CMD $CMD

创建训练作业 方式一：使用PyTorch预置框架功能，通过mp.spawn命令启动训练作业。 创建训练作业的关键参数如表1所示。 表1 创建训练作业（预置框架） 参数名称 说明 创建方式 选择“自定义算法”。 启动方式 选择“预置框架”，引擎选择“PyTorch”，PyTorch版本根据训练要求选择。 代码目录 选择OBS桶中训练code文件夹所在路径，例如“obs://test-modelarts/code/”。 启动文件 选择代码目录中训练作业的Python启动脚本。例如“obs://test-modelarts/code/main.py”。 超参 当资源规格为单机多卡时，需要指定超参world_size和rank。 当资源规格为多机时（即计算节点个数大于 1），无需设置超参world_size和rank，超参会由平台自动注入。 方式二：使用自定义镜像功能，通过torch.distributed.launch命令启动训练作业。 创建训练作业的关键参数如表2所示。 表2 创建训练作业（自定义镜像+torch.distributed.launch命令） 参数名称 说明 创建方式 选择“自定义算法”。 启动方式 选择“自定义”。 镜像 选择用于训练的PyTorch镜像。 代码目录 选择OBS桶中训练code文件夹所在路径，例如“obs://test-modelarts/code/”。 启动命令 输入镜像的Python启动命令，例如： bash ${MA_JOB_DIR}/code/torchlaunch.sh 方式三：使用自定义镜像功能，通过torch.distributed.run命令启动训练作业。 创建训练作业的关键参数如表3所示。 表3 创建训练作业（自定义镜像+torch.distributed.run命令） 参数名称 说明 创建方式 选择“自定义算法”。 启动方式 选择“自定义”。 镜像 选择用于训练的PyTorch镜像。 代码目录 选择OBS桶中训练code文件夹所在路径，例如“obs://test-modelarts/code/”。 启动命令 输入镜像的Python启动命令，例如： bash ${MA_JOB_DIR}/code/torchrun.sh

数据集 cifar10数据集 在Notebook中，无法直接使用默认版本的torchvision获取数据集，因此示例代码中提供了三种训练数据加载方式。 cifar-10数据集下载链接，单击“CIFAR-10 python version”。 尝试基于torchvision获取cifar10数据集。 基于数据链接下载数据并解压，放置在指定目录下，训练集和测试集的大小分别为(50000，3，32，32)和(10000，3，32，32)。 考虑到下载cifar10数据集较慢，基于torch生成类似cifar10的随机数据集，训练集和测试集的大小分别为(5000，3，32，32)和(1000，3，32，32)，标签仍为10类，指定custom_data = 'true'后可直接进行训练任务，无需加载数据。

训练代码 以下代码中以“### 分布式改造，... ###”注释的代码即为多节点分布式训练需要适配的代码改造点。 不对示例代码进行任何修改，适配数据路径后即可在ModelArts上完成多节点分布式训练。 注释掉分布式代码改造点，即可完成单节点单卡训练。完整代码见分布式训练完整代码示例。 导入依赖包 import datetime import inspect import os import pickle import random import argparse import numpy as np import torch import torch.distributed as dist from torch import nn, optim from torch.utils.data import TensorDataset, DataLoader from torch.utils.data.distributed import DistributedSampler from sklearn.metrics import accuracy_score 定义加载数据的方法和随机数，由于加载数据部分代码较多，此处省略 def setup_seed(seed): torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) np.random.seed(seed) random.seed(seed) torch.backends.cudnn.deterministic = True def get_data(path): pass 定义网络结构 class Block(nn.Module): def __init__(self, in_channels, out_channels, stride=1): super().__init__() self.residual_function = nn.Sequential( nn.Conv2d(in_channels, out_channels, kernel_size=3, stride=stride, padding=1, bias=False), nn.BatchNorm2d(out_channels), nn.ReLU(inplace=True), nn.Conv2d(out_channels, out_channels, kernel_size=3, padding=1, bias=False), nn.BatchNorm2d(out_channels) ) self.shortcut = nn.Sequential() if stride != 1 or in_channels != out_channels: self.shortcut = nn.Sequential( nn.Conv2d(in_channels, out_channels, kernel_size=1, stride=stride, bias=False), nn.BatchNorm2d(out_channels) ) def forward(self, x): out = self.residual_function(x) + self.shortcut(x) return nn.ReLU(inplace=True)(out) class ResNet(nn.Module): def __init__(self, block, num_classes=10): super().__init__() self.conv1 = nn.Sequential( nn.Conv2d(3, 64, kernel_size=3, padding=1, bias=False), nn.BatchNorm2d(64), nn.ReLU(inplace=True)) self.conv2 = self.make_layer(block, 64, 64, 2, 1) self.conv3 = self.make_layer(block, 64, 128, 2, 2) self.conv4 = self.make_layer(block, 128, 256, 2, 2) self.conv5 = self.make_layer(block, 256, 512, 2, 2) self.avg_pool = nn.AdaptiveAvgPool2d((1, 1)) self.dense_layer = nn.Linear(512, num_classes) def make_layer(self, block, in_channels, out_channels, num_blocks, stride): strides = [stride] + [1] * (num_blocks - 1) layers = [] for stride in strides: layers.append(block(in_channels, out_channels, stride)) in_channels = out_channels return nn.Sequential(*layers) def forward(self, x): out = self.conv1(x) out = self.conv2(out) out = self.conv3(out) out = self.conv4(out) out = self.conv5(out) out = self.avg_pool(out) out = out.view(out.size(0), -1) out = self.dense_layer(out) return out 进行训练和验证 def main(): file_dir = os.path.dirname(inspect.getframeinfo(inspect.currentframe()).filename) seed = datetime.datetime.now().year setup_seed(seed) parser = argparse.ArgumentParser(description='Pytorch distribute training', formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument('--enable_gpu', default='true') parser.add_argument('--lr', default='0.01', help='learning rate') parser.add_argument('--epochs', default='100', help='training iteration') parser.add_argument('--init_method', default=None, help='tcp_port') parser.add_argument('--rank', type=int, default=0, help='index of current task') parser.add_argument('--world_size', type=int, default=1, help='total number of tasks') parser.add_argument('--custom_data', default='false') parser.add_argument('--data_url', type=str, default=os.path.join(file_dir, 'input_dir')) parser.add_argument('--output_dir', type=str, default=os.path.join(file_dir, 'output_dir')) args, unknown = parser.parse_known_args() args.enable_gpu = args.enable_gpu == 'true' args.custom_data = args.custom_data == 'true' args.lr = float(args.lr) args.epochs = int(args.epochs) if args.custom_data: print('[warning] you are training on custom random dataset, ' 'validation accuracy may range from 0.4 to 0.6.') ### 分布式改造，DDP初始化进程，其中init_method, rank和world_size参数均由平台自动入参 ### dist.init_process_group(init_method=args.init_method, backend="nccl", world_size=args.world_size, rank=args.rank) ### 分布式改造，DDP初始化进程，其中init_method, rank和world_size参数均由平台自动入参 ### tr_set, val_set = get_data(args.data_url, custom_data=args.custom_data) batch_per_gpu = 128 gpus_per_node = torch.cuda.device_count() if args.enable_gpu else 1 batch = batch_per_gpu * gpus_per_node tr_loader = DataLoader(tr_set, batch_size=batch, shuffle=False) ### 分布式改造，构建DDP分布式数据sampler，确保不同进程加载到不同的数据 ### tr_sampler = DistributedSampler(tr_set, num_replicas=args.world_size, rank=args.rank) tr_loader = DataLoader(tr_set, batch_size=batch, sampler=tr_sampler, shuffle=False, drop_last=True) ### 分布式改造，构建DDP分布式数据sampler，确保不同进程加载到不同的数据 ### val_loader = DataLoader(val_set, batch_size=batch, shuffle=False) lr = args.lr * gpus_per_node max_epoch = args.epochs model = ResNet(Block).cuda() if args.enable_gpu else ResNet(Block) ### 分布式改造，构建DDP分布式模型 ### model = nn.parallel.DistributedDataParallel(model) ### 分布式改造，构建DDP分布式模型 ### optimizer = optim.Adam(model.parameters(), lr=lr) loss_func = torch.nn.CrossEntropyLoss() os.makedirs(args.output_dir, exist_ok=True) for epoch in range(1, max_epoch + 1): model.train() train_loss = 0 ### 分布式改造，DDP sampler, 基于当前的epoch为其设置随机数，避免加载到重复数据 ### tr_sampler.set_epoch(epoch) ### 分布式改造，DDP sampler, 基于当前的epoch为其设置随机数，避免加载到重复数据 ### for step, (tr_x, tr_y) in enumerate(tr_loader): if args.enable_gpu: tr_x, tr_y = tr_x.cuda(), tr_y.cuda() out = model(tr_x) loss = loss_func(out, tr_y) optimizer.zero_grad() loss.backward() optimizer.step() train_loss += loss.item() print('train | epoch: %d | loss: %.4f' % (epoch, train_loss / len(tr_loader))) val_loss = 0 pred_record = [] real_record = [] model.eval() with torch.no_grad(): for step, (val_x, val_y) in enumerate(val_loader): if args.enable_gpu: val_x, val_y = val_x.cuda(), val_y.cuda() out = model(val_x) pred_record += list(np.argmax(out.cpu().numpy(), axis=1)) real_record += list(val_y.cpu().numpy()) val_loss += loss_func(out, val_y).item() val_accu = accuracy_score(real_record, pred_record) print('val | epoch: %d | loss: %.4f | accuracy: %.4f' % (epoch, val_loss / len(val_loader), val_accu), '\n') if args.rank == 0: # save ckpt every epoch torch.save(model.state_dict(), os.path.join(args.output_dir, f'epoch_{epoch}.pth')) if __name__ == '__main__': main() 结果对比 分别以单机单卡和两节点16卡两种资源类型完成100epoch的cifar-10数据集训练，训练时长和测试集准确率如下。 表1 训练结果对比 资源类型 单机单卡 两节点16卡 耗时 60分钟 20分钟 准确率 80+ 80+

训练流程简述 相比于DP，DDP能够启动多进程进行运算，从而大幅度提升计算资源的利用率。可以基于torch.distributed实现真正的分布式计算，具体的原理此处不再赘述。大致的流程如下： 初始化进程组。 创建分布式并行模型，每个进程都会有相同的模型和参数。 创建数据分发Sampler，使每个进程加载一个mini batch中不同部分的数据。 网络中相邻参数分桶，一般为神经网络模型中需要进行参数更新的每一层网络。 每个进程前向传播并各自计算梯度。 模型某一层的参数得到梯度后会马上进行通讯并进行梯度平均。 各GPU更新模型参数。 具体流程图如下： 图1 多机多卡数据并行训练