本系列将利用阿里云容器服务,帮助您上手Kubeflow Pipelines.python
从上篇文章中,咱们能够看到如何经过Kubeflow Pipeline运行单节点任务机器学习工做流,在本文中,咱们会介绍如何使用Pipeline运行分布式MPI任务,该MPI任务运行模型ResNet101的测试。git
因为Kubeflow Pipelines提供的例子多数都是单机任务,那么如何利用Pipelines运行分布式训练?阿里云容器服务团队提供了利用MPIJob训练ResNet101模型的例子,方便您在阿里云上使用和学习Kubeflow Pipelines,而且训练支持分布式的allreduce模型训练。github
在Kubeflow Pipelines中能够用Python代码描述了这样一个流程, 完整代码能够查看mpi_run.py。咱们在这个例子中使用了arena_op
这是对于Kubeflow默认的container_op
封装,可以实现对于分布式训练MPI和PS模式的无缝衔接,另外也支持使用GPU和RDMA等异构设备和分布式存储的无缝接入,同时也方便从git源同步代码。是一个比较实用的工具API。而arena_op
是基于开源项目Arena。web
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
函数的参数以下:docker
name,image, data和command以外,在模型训练步骤中,还须要指定:数组
[]
metrics=["Train-accuracy:PERCENTAGE"]
。2.因为Pipelines默认会从stdout日志中收集指标,你须要在真正运行的模型代码中输出{metrics name}={value}或者{metrics name}:{value}, 能够参考具体样例代码arena中mpi_run方法的调用, 还比较直观,下面就能够定义一个Python方法。bash
@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
curl入口方法
mpirun_pipeline
中定义了一系列参数,因为数量较多就不在这里一一列举了。这里的参数的值其实是 dsl.PipelineParam类型,定义成dsl.PipelineParam的目的在于能够经过Kubeflow Pipelines的原生UI能够将其转换成输入表单,表单的关键字是参数名称,而默认值为参数的值. 值得注意的是,这里的dsl.PipelineParam对应值的实际上只能是字符串和数字型;而数组和map,以及自定义类型都是没法经过转型进行变换的。机器学习
而实际上,这些参数均可以在用户提交工做流时进行覆盖,如下就是提交工做流对应的UI:分布式
整个过程包括:
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-beijing.aliyuncs.com/kfp/0.1.16/kfp.tar.gz --upgrade # pip3 install http://kubeflow.oss-cn-beijing.aliyuncs.com/kfp-arena/kfp-arena-0.6.tar.gz --upgrade # curl -O https://raw.githubusercontent.com/cheyang/pipelines/add_mpijob/samples/arena-samples/mpi/mpi_run.py # 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的输入参数,也是有足够的灵活性。
原文连接 本文为云栖社区原创内容,未经容许不得转载。