devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

分布式深度学习框架--百度paddlepaddle的环境搭建以及测试代码_深度学习分布式测试

作者：笔触狂放9 | 2024-04-04 22:20:41

赞

踩

深度学习分布式测试

一、搭建环境

      分布式paddle的环境和tensorflow的环境很相似，在我之前的博客中已经写得非常清楚了

      我的是 python3.6+cuda9+cudnn7+ubuntu16.04

      博客链接：https://blog.csdn.net/qq_28626909/article/details/85007363

      搭建好环境知道我们开始安装paddlepaddle，

      官方网站链接：http://www.paddlepaddle.org/

      

    我的配置是NVIDIA384+cuda9.0+cudnn7.0,所以我直接pip install paddlepaddle-gpu即可

     cuda8+cudnn7.0是pip install paddlepaddle-gpu==1.2.0.post87

     cuda8+cudnn5.0是pip install paddlepaddle-gpu==1.2.0.post85

二，测试单机paddle

    为了测试我们的paddle是否安装成功，我们这里放出代码，可以直接跑的。

  1. #!D:/workplace/python
  2. # -*- coding: utf-8 -*-
  3. # @File  : first_test.py
  4. # @Author: WangYe
  5. # @Date  : 2019/1/10
  6. # @Software: PyCharm
  7. import paddle
  8. import paddle.fluid as fluid
  9. import numpy
  10. import six
  11.  
  12.  
  13. # Configure the neural network.
  14. def net(x, y):
  15.     y_predict = fluid.layers.fc(input=x, size=1, act=None)
  16.     cost = fluid.layers.square_error_cost(input=y_predict, label=y)
  17.     avg_cost = fluid.layers.mean(cost)
  18.     return y_predict, avg_cost
  19.  
  20.  
  21. # Define train function.
  22. def train(save_dirname):
  23.     x = fluid.layers.data(name='x', shape=[13], dtype='float32')
  24.     y = fluid.layers.data(name='y', shape=[1], dtype='float32')
  25.     y_predict, avg_cost = net(x, y)
  26.     sgd_optimizer = fluid.optimizer.SGD(learning_rate=0.001)
  27.     sgd_optimizer.minimize(avg_cost)
  28.     train_reader = paddle.batch(
  29.         paddle.reader.shuffle(paddle.dataset.uci_housing.train(), buf_size=500),
  30.         batch_size=20)
  31.     place = fluid.CPUPlace()
  32.     exe = fluid.Executor(place)
  33.  
  34.     def train_loop(main_program):
  35.         feeder = fluid.DataFeeder(place=place, feed_list=[x, y])
  36.         exe.run(fluid.default_startup_program())
  37.  
  38.         PASS_NUM = 1000
  39.         for pass_id in range(PASS_NUM):
  40.             total_loss_pass = 0
  41.             for data in train_reader():
  42.                 avg_loss_value, = exe.run(
  43.                     main_program, feed=feeder.feed(data), fetch_list=[avg_cost])
  44.                 total_loss_pass += avg_loss_value
  45.                 if avg_loss_value < 5.0:
  46.                     if save_dirname is not None:
  47.                         fluid.io.save_inference_model(
  48.                             save_dirname, ['x'], [y_predict], exe)
  49.                     return
  50.             print("Pass %d, total avg cost = %f" % (pass_id, total_loss_pass))
  51.  
  52.     train_loop(fluid.default_main_program())
  53.  
  54.  
  55. # Infer by using provided test data.
  56. def infer(save_dirname=None):
  57.     place = fluid.CPUPlace()
  58.     exe = fluid.Executor(place)
  59.     inference_scope = fluid.core.Scope()
  60.     with fluid.scope_guard(inference_scope):
  61.         [inference_program, feed_target_names, fetch_targets] = (
  62.             fluid.io.load_inference_model(save_dirname, exe))
  63.         test_reader = paddle.batch(paddle.dataset.uci_housing.test(), batch_size=20)
  64.  
  65.         test_data = six.next(test_reader())
  66.         test_feat = numpy.array(list(map(lambda x: x[0], test_data))).astype("float32")
  67.         test_label = numpy.array(list(map(lambda x: x[1], test_data))).astype("float32")
  68.  
  69.         results = exe.run(inference_program,
  70.                           feed={feed_target_names[0]: numpy.array(test_feat)},
  71.                           fetch_list=fetch_targets)
  72.         print("infer results: ", results[0])
  73.         print("ground truth: ", test_label)
  74.  
  75.  
  76. # Run train and infer.
  77. if __name__ == "__main__":
  78.     save_dirname = "fit_a_line.inference.model"
  79.     train(save_dirname)
  80.     infer(save_dirname)

代码的很多部分应该用过tensorflow的人都能看懂，paddlepaddle还是比tensorflow更简洁的   

运行结果截图：

  三、分布式paddle的测试

      分布式paddle和tensorflow是有很多相似之处，我这里对官方网站的代码进行了一点改动，这样测试起来就更为方便

   tensorflow把参数服务器叫ps，计算节点叫worker

   padddle把参数服务器叫PSERVER,计算节点叫TRAINER

    这是第一段，也就是PSERVER的代码，我把代码命名为：

   dist_train_ps.py

  1. #!D:/workplace/python
  2. # -*- coding: utf-8 -*-
  3. # @File  : dist_train_ps.py
  4. # @Author: WangYe
  5. # @Date  : 2019/1/11
  6. # @Software: PyCharm
  7.  
  8. import os
  9. import paddle
  10. import paddle.fluid as fluid
  11.  
  12. # train reader
  13. BATCH_SIZE = 20
  14. EPOCH_NUM = 30
  15. BATCH_SIZE = 8
  16.  
  17. train_reader = paddle.batch(
  18.     paddle.reader.shuffle(
  19.         paddle.dataset.uci_housing.train(), buf_size=500),
  20.     batch_size=BATCH_SIZE)
  21.  
  22. def train():
  23.     y = fluid.layers.data(name='y', shape=[1], dtype='float32')
  24.     x = fluid.layers.data(name='x', shape=[13], dtype='float32')
  25.     y_predict = fluid.layers.fc(input=x, size=1, act=None)
  26.  
  27.     loss = fluid.layers.square_error_cost(input=y_predict, label=y)
  28.     avg_loss = fluid.layers.mean(loss)
  29.     opt = fluid.optimizer.SGD(learning_rate=0.001)
  30.     opt.minimize(avg_loss)
  31.  
  32.     place = fluid.CPUPlace()
  33.     feeder = fluid.DataFeeder(place=place, feed_list=[x, y])
  34.     exe = fluid.Executor(place)
  35.  
  36.     # fetch distributed training environment setting
  37.     training_role = os.getenv("PADDLE_TRAINING_ROLE", "PSERVER")
  38.     port = os.getenv("PADDLE_PSERVER_PORT", "6174")
  39.     pserver_ips = os.getenv("PADDLE_PSERVER_IPS", "localhost")
  40.     trainer_id = int(os.getenv("PADDLE_TRAINER_ID", "0"))
  41.     eplist = []
  42.     for ip in pserver_ips.split(","):
  43.         eplist.append(':'.join([ip, port]))
  44.     pserver_endpoints = ",".join(eplist)
  45.     trainers = int(os.getenv("PADDLE_TRAINERS","1"))
  46.     current_endpoint = os.getenv("PADDLE_CURRENT_IP", "localhost") + ":" + port
  47.  
  48.     t = fluid.DistributeTranspiler()
  49.     t.transpile(
  50.         trainer_id = trainer_id,
  51.         pservers = pserver_endpoints,
  52.         trainers = trainers)
  53.  
  54.     if training_role == "PSERVER":
  55.         pserver_prog = t.get_pserver_program(current_endpoint)
  56.         startup_prog = t.get_startup_program(current_endpoint, pserver_prog)
  57.         exe.run(startup_prog)
  58.         exe.run(pserver_prog)
  59.     elif training_role == "TRAINER":
  60.         trainer_prog = t.get_trainer_program()
  61.         exe.run(fluid.default_startup_program())
  62.  
  63.         for epoch in range(EPOCH_NUM):
  64.             for batch_id, batch_data in enumerate(train_reader()):
  65.                 avg_loss_value, = exe.run(trainer_prog,
  66.                                       feed=feeder.feed(batch_data),
  67.                                       fetch_list=[avg_loss])
  68.                 if (batch_id + 1) % 10 == 0:
  69.                     print("Epoch: {0}, Batch: {1}, loss: {2}".format(
  70.                         epoch, batch_id, avg_loss_value[0]))
  71.         # destory the resource of current trainer node in pserver server node
  72.         exe.close()
  73.     else:
  74.         raise AssertionError("PADDLE_TRAINING_ROLE should be one of [TRAINER, PSERVER]")
  75.  
  76. train()

    下面的是第二段，也就是的TRAINER，我把代码命名为

   dist_train_worker.py

  1. #!D:/workplace/python
  2. # -*- coding: utf-8 -*-
  3. # @File  : dist_train_ps.py
  4. # @Author: WangYe
  5. # @Date  : 2019/1/11
  6. # @Software: PyCharm
  7.  
  8. import os
  9. import paddle
  10. import paddle.fluid as fluid
  11.  
  12. # train reader
  13. BATCH_SIZE = 20
  14. EPOCH_NUM = 30
  15. BATCH_SIZE = 8
  16.  
  17. train_reader = paddle.batch(
  18.     paddle.reader.shuffle(
  19.         paddle.dataset.uci_housing.train(), buf_size=500),
  20.     batch_size=BATCH_SIZE)
  21.  
  22. def train():
  23.     y = fluid.layers.data(name='y', shape=[1], dtype='float32')
  24.     x = fluid.layers.data(name='x', shape=[13], dtype='float32')
  25.     y_predict = fluid.layers.fc(input=x, size=1, act=None)
  26.  
  27.     loss = fluid.layers.square_error_cost(input=y_predict, label=y)
  28.     avg_loss = fluid.layers.mean(loss)
  29.     opt = fluid.optimizer.SGD(learning_rate=0.001)
  30.     opt.minimize(avg_loss)
  31.  
  32.     place = fluid.CPUPlace()
  33.     feeder = fluid.DataFeeder(place=place, feed_list=[x, y])
  34.     exe = fluid.Executor(place)
  35.  
  36.     # fetch distributed training environment setting
  37.     training_role = os.getenv("PADDLE_TRAINING_ROLE", "TRAINER")
  38.     port = os.getenv("PADDLE_PSERVER_PORT", "6174")
  39.     pserver_ips = os.getenv("PADDLE_PSERVER_IPS", "localhost")
  40.     trainer_id = int(os.getenv("PADDLE_TRAINER_ID", "0"))
  41.     eplist = []
  42.     for ip in pserver_ips.split(","):
  43.         eplist.append(':'.join([ip, port]))
  44.     pserver_endpoints = ",".join(eplist)
  45.     trainers = int(os.getenv("PADDLE_TRAINERS","1"))
  46.     current_endpoint = os.getenv("PADDLE_CURRENT_IP", "") + ":" + port
  47.  
  48.     t = fluid.DistributeTranspiler()
  49.     t.transpile(
  50.         trainer_id = trainer_id,
  51.         pservers = pserver_endpoints,
  52.         trainers = trainers)
  53.  
  54.     if training_role == "PSERVER":
  55.         pserver_prog = t.get_pserver_program(current_endpoint)
  56.         startup_prog = t.get_startup_program(current_endpoint, pserver_prog)
  57.         exe.run(startup_prog)
  58.         exe.run(pserver_prog)
  59.     elif training_role == "TRAINER":
  60.         trainer_prog = t.get_trainer_program()
  61.         exe.run(fluid.default_startup_program())
  62.  
  63.         for epoch in range(EPOCH_NUM):
  64.             for batch_id, batch_data in enumerate(train_reader()):
  65.                 avg_loss_value, = exe.run(trainer_prog,
  66.                                       feed=feeder.feed(batch_data),
  67.                                       fetch_list=[avg_loss])
  68.                 if (batch_id + 1) % 10 == 0:
  69.                     print("Epoch: {0}, Batch: {1}, loss: {2}".format(
  70.                         epoch, batch_id, avg_loss_value[0]))
  71.         # destory the resource of current trainer node in pserver server node
  72.         exe.close()
  73.     else:
  74.         raise AssertionError("PADDLE_TRAINING_ROLE should be one of [TRAINER, PSERVER]")
  75.  
  76. train()

对于这两端代码，有一下几点说明

1.分布式测试代码用到的数据是上面第二部分单机测试的数据，它的数据是自动下载的。也就是说运行过上面的单机代码后，就自动有了分布式的数据

2.分布式代码需要分开执行，同tensorflow一样，不过我做的改动在代码里了，所以执行代码之后命令行不用添加参数，如下截图

 

 

 

3.可能遇到的问题：

  除开我刚开始放出的链接中遇到的环境问题，我遇到了一个 段错误的一个问题，那是因为ps和worker的ip写的有问题

四、分布式paddle的关键参数

  

  1. training_role = os.getenv("PADDLE_TRAINING_ROLE", "TRAINER")
  2.     port = os.getenv("PADDLE_PSERVER_PORT", "6174")
  3.     pserver_ips = os.getenv("PADDLE_PSERVER_IPS", "localhost")
  4.     trainer_id = int(os.getenv("PADDLE_TRAINER_ID", "0"))
  5.     eplist = []
  6.     for ip in pserver_ips.split(","):
  7.         eplist.append(':'.join([ip, port]))
  8.     pserver_endpoints = ",".join(eplist)
  9.     trainers = int(os.getenv("PADDLE_TRAINERS","1"))
  10.     current_endpoint = os.getenv("PADDLE_CURRENT_IP", "localhost") + ":" + port

  我们一个一个来看，以上面的worker的代码为例（paddle中把worker成为tariner）,从上往下讲

  paddle_tarining_role（str）：训练对象（trainer还是pserver）

 paddle_pserver_port(int):端口（这里和tensorflow不同的是必须指定同一个端口，即所有tariner和ps都是6174）

 paddle_pserver_ips(str):ps的ip地址，如果多个就用逗号隔开

 paddle_trainer_id(int):tariner各自的编号，在ps中全部填写0

 paddle_trainers(int):trainer的数量，即tensorflow中多少个worker

paddle_current_ip(str):当前ps节点的ip地址，如果是worker，则可以不指定

官方网站链接：http://www.paddlepaddle.org/documentation/docs/zh/1.2/user_guides/howto/training/cluster_quick_start.html

 官方参数解读网站截图：

 

 运行截图：

 

ok，大功告成！！

 

 

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号