像Google同样构建机器学习系统3 - 利用MPIJob运行ResNet101

摘要： 从上篇文章中，咱们能够看到如何经过Kubeflow Pipeline运行单节点任务机器学习工做流，在本文中，咱们会介绍如何使用Pipeline运行分布式MPI任务，该MPI任务运行模型ResNet101的测试。

本系列将利用阿里云容器服务，帮助您上手Kubeflow Pipelines.

第一篇：在阿里云上搭建Kubeflow Pipelines
第二篇：开发你的机器学习工做流
第三篇：利用MPIJob运行ResNet101
从上篇文章中，咱们能够看到如何经过Kubeflow Pipeline运行单节点任务机器学习工做流，在本文中，咱们会介绍如何使用Pipeline运行分布式MPI任务，该MPI任务运行模型ResNet101的测试。

开发MPIRun Pipeline
因为Kubeflow Pipelines提供的例子多数都是单机任务，那么如何利用Pipelines运行分布式训练？阿里云容器服务团队提供了利用MPIJob训练ResNet101模型的例子，方便您在阿里云上使用和学习Kubeflow Pipelines，而且训练支持分布式的allreduce模型训练。

在Kubeflow Pipelines中能够用Python代码描述了这样一个流程, 完整代码能够查看mpi_run.py。咱们在这个例子中使用了arena_op这是对于Kubeflow默认的container_op封装，可以实现对于分布式训练MPI和PS模式的无缝衔接，另外也支持使用GPU和RDMA等异构设备和分布式存储的无缝接入，同时也方便从git源同步代码。是一个比较实用的工具API。而arena_op是基于开源项目Arena。

env = ['NCCL_DEBUG=INFO','GIT_SYNC_BRANCH={0}'.format(git_sync_branch)]

train=arena.mpi_job_op(

name="all-reduce",
image=image,
env=env,
data=[data],
workers=workers,
sync_source=sync_source,
gpus=gpus,
cpu_limit=cpu_limit,
memory_limit=memory_limit,
metrics=[metric],
command="""
mpirun python code/benchmarks/scripts/tf_cnn_benchmarks/tf_cnn_benchmarks.py --model resnet101 \
--batch_size {0}  --variable_update horovod --optimizer {1}\
--summary_verbosity=3 --save_summaries_steps=10
""".format(batch_size, optimizer)

)
arena.mpi_job_op函数的参数以下:

name，image, data和command以外，在模型训练步骤中，还须要指定:

name: 步骤名称
image: 须要使用的容器镜像
workers: 参与运算的worker数量
data: 要使用的数据以及其对应到容器内部的挂载目录, 这里的data是一个数组类型, 能够设置为data=["user-susan:/training"]，表示能够挂载到多个数据。
env: 系统环境变量，这里的env也是数组类型，能够支持多个env; 若是不须要指定环境变量能够把env设置为[]
gpu: 默认为0，就是不使用GPU；若是为大于0的整数值，就表明该步骤须要这个数量的GPU数。
cpu_limit: cpu的资源上限
memory_limit: 内存的资源上限
metrics: 一样是从可重现和可比较的实验目的出发，用户能够将须要的一系列指标导出，而且经过Pipelines UI上直观的显示和比较。具体使用方法分为两步，1.在调用API时以数组的形式指定要收集指标的metrics name和指标的展现格式PERCENTAGE或者是RAW，好比metrics=["Train-accuracy:PERCENTAGE"]。2.因为Pipelines默认会从stdout日志中收集指标，你须要在真正运行的模型代码中输出{metrics name}={value}或者{metrics name}:{value}, 能够参考具体样例代码
command: 就是要提交的mpirun命令
arena中mpi_run方法的调用, 还比较直观，下面就能够定义一个Python方法。

@dsl.pipeline(
name='pipeline to run mpi job',
description='shows how to run mpi job.'
)
def mpirun_pipeline(image="uber/horovod:0.13.11-tf1.10.0-torch0.4.0-py3.5",

batch_size="64",
           optimizer='momentum',
           sync_source='https://github.com/tensorflow/benchmarks.git',
           git_sync_branch='cnn_tf_v1.9_compatible',
           data='user-susan:/training',
           gpus=1,
           workers=1,
           cpu_limit='2',
           metric='images/sec',
           memory_limit='10Gi'):

@dsl.pipeline是表示工做流的装饰器，这个装饰器中须要定义两个属性，分别是name和description

入口方法mpirun_pipeline中定义了一系列参数，因为数量较多就不在这里一一列举了。这里的参数的值其实是 dsl.PipelineParam类型，定义成dsl.PipelineParam的目的在于能够经过Kubeflow Pipelines的原生UI能够将其转换成输入表单，表单的关键字是参数名称，而默认值为参数的值. 值得注意的是，这里的dsl.PipelineParam对应值的实际上只能是字符串和数字型；而数组和map，以及自定义类型都是没法经过转型进行变换的。

而实际上，这些参数均可以在用户提交工做流时进行覆盖，如下就是提交工做流对应的UI：

提交Pipeline
整个过程包括：

1.将Python代码编译成Pipelines执行引擎(Argo)识别的DAG文件的压缩包

准备一个python3的环境，而且安装Kubeflow Pipelines SDK

docker run -itd --name py3 python:3 sleep infinity

docker exec -it py3 bash

在Python3的环境下执行以下命令

pip3 install http://kubeflow.oss-cn-beijin... --upgrade

pip3 install http://kubeflow.oss-cn-beijin... --upgrade

curl -O https://raw.githubusercontent...

dsl-compile --py mpi_run.py --output mpi_run.py.tar.gz

ls -ltr | grep mpi_run

mpi_run.py.tar.gz

exit

将该文件从容器中拷贝出来

docker cp py3:/mpi_run.py.tar.gz .

2.将该压缩包上传到Kubeflow Pipeline的web控制台，而且将名字改成mpi_run

运行试验
1.在pipeline页面，点击mpi_run连接

2.点击右上角按钮Create run

3.在Start a new run的界面上填写Run name，同时选择已有或者建立相关的实验。同时按照实际状况设置运行参数，也就是Run parameters。注意，若是您没有配置数据相关的配置，请将data中的参数清空便可。点击启动便可。

查看运行结果
登陆到Kubeflow Pipelines的UI: [https://{pipeline地址}/pipeline/#/experiments]，查看实验结果：

点击具体Run，选择all-reduce, 并点击logs查看日志

总结
本文介绍了如何利用Pipeline运行MPIJob，实际上这个例子并不必定严丝合缝的知足使用者的需求：

arena.mpi_job_op是使用MPIJob的API，您能够根据须要自定义调用方式。def mpirun_pipeline的灵活性更是掌握在用户的手里，如何定义具体Pipeline的输入参数，也是有足够的灵活性。