調(diào)參俠看過來！兩個(gè)提高深度學(xué)習(xí)訓(xùn)練效率的絕技

1. 訓(xùn)練的瓶頸在哪里

• GPU利用率低：模型訓(xùn)練時(shí)GPU顯存沾滿了，但是GPU的利用率比較不穩(wěn)定，有時(shí)候0%，有時(shí)候90%，忽高忽低。

• 訓(xùn)練的數(shù)據(jù)量大：訓(xùn)練數(shù)據(jù)大，在百萬/千萬的量級，訓(xùn)練一個(gè)Epoch需要很長時(shí)間，模型迭代周期過長。

2. 提高GPU利用率：CPU vs GPU

GPU利用率低, 主要原因是CPU處理的效率跟不上GPU

2.1 CPU vs GPU的通信

• CPU負(fù)責(zé)加載數(shù)據(jù)+數(shù)據(jù)預(yù)處理，并不斷的在內(nèi)存和顯存之間交互數(shù)據(jù)
• GPU負(fù)責(zé)模型訓(xùn)練（圖片來自網(wǎng)絡(luò)）

2.2 解決方案

采用多進(jìn)程并行處理，加快CPU加載數(shù)據(jù)的性能

• keras keras 中提供了workers use_multiprocessing來采用多進(jìn)程方式，并行處理數(shù)據(jù)，并push到隊(duì)列中，共GPU模型訓(xùn)練。因?yàn)檫M(jìn)程之間可能相互影響資源，并不是越大越好，workers可以設(shè)置2，4，8。

  run_model.fit_generator(
  ????????????generator=training_generator,
  ????????????class_weight={0:?config.weights,?1:?1},
  ????????????epochs=epochs,
  ????????????verbose=1,
  ????????????steps_per_epoch=steps_per_epoch,
  ????????????callbacks=callbacks_list,
  ????????????validation_data=valid_generator,
  ????????????validation_steps=validation_steps,
  ????????????shuffle=True,
  ????????????workers=8,
  ????????????use_multiprocessing=True,
  ????????????max_queue_size=20
  

• pytorch torch在加載數(shù)據(jù)中提供類似參數(shù)num_workers。pin_memory=True可以直接加載到顯存中，而不需要內(nèi)存

  torch.utils.data.DataLoader(image_datasets[x],
  ????????????????????????????batch_size=batch_size,?
  ????????????????????????????shuffle=True,
  ????????????????????????????num_workers=8,
  ????????????????????????????pin_memory=True)
  

3. 分布式并行訓(xùn)練

3.1 并行模式

當(dāng)訓(xùn)練的數(shù)據(jù)量很大時(shí)，可以通過多個(gè)機(jī)器多個(gè)GPU來提高訓(xùn)練的效率。不同于hadoop和spark等分布式數(shù)據(jù)處理框架，深度學(xué)習(xí)訓(xùn)練因?yàn)橐婕皡?shù)的前項(xiàng)傳播和反向傳播，有兩種并行方式：

• 模型并行（ model parallelism ）:分布式系統(tǒng)中的不同機(jī)器（GPU/CPU等）負(fù)責(zé)網(wǎng)絡(luò)模型的不同部分，通常是神經(jīng)網(wǎng)絡(luò)模型的不同網(wǎng)絡(luò)層被分配到不同的機(jī)器，或者同一層內(nèi)部的不同參數(shù)被分配到不同機(jī)器。一般是超大的模型，一張顯卡放不下的情況，如NLP的模型。模型并行的缺點(diǎn)是層和層之間可能存在依賴關(guān)系，不能完全的并行。（圖片來自網(wǎng)絡(luò)）

• 數(shù)據(jù)并行（ data parallelism ）：不同的機(jī)器有同一個(gè)模型的多個(gè)副本，每個(gè)機(jī)器分配到不同的數(shù)據(jù)，然后將所有機(jī)器的計(jì)算結(jié)果按照某種方式合并。這種就比較適合大數(shù)據(jù)的情況。數(shù)據(jù)并行要解決的問題是數(shù)據(jù)的分割和傳輸，以及參數(shù)的更新。

3.2 數(shù)據(jù)并行

Facebook在《Accurate, Large Minibatch SGD: Training ImageNet in 1 Hour》介紹了使用 256 塊 GPU 進(jìn)行 ResNet-50 網(wǎng)絡(luò)「數(shù)據(jù)并行」訓(xùn)練的方法

• 數(shù)據(jù)分割: 選用大的batch-size, 按照worker數(shù)量進(jìn)行分割， 分發(fā)到不同worker執(zhí)行
• 參數(shù)更新：參數(shù)的更新有兩種模式（1）參數(shù)服務(wù)器 （2） ring環(huán)狀更新（無服務(wù)器模式）

3.2.1 參數(shù)服務(wù)器模式

參數(shù)服務(wù)器模式，見下圖。在每個(gè)worker執(zhí)行完一個(gè)batch的訓(xùn)練后，反向傳播參數(shù)的時(shí)候，所有的worker都會(huì)把參數(shù)傳給參數(shù)服務(wù)器，進(jìn)行匯總求均值，之后再傳給每個(gè)worker，進(jìn)入第二個(gè)batch的訓(xùn)練。（圖片來自網(wǎng)絡(luò)）

參數(shù)服務(wù)器有一個(gè)或者多個(gè)的結(jié)構(gòu)模式，可以看出這種數(shù)據(jù)并行的模式效率是否提升取決于參數(shù)服務(wù)器與worker之間的通信效率，也就是最慢的worker的訓(xùn)練時(shí)間和參數(shù)服務(wù)器的接收和更新參數(shù)后再回傳的時(shí)間。worker數(shù)量多的話，參數(shù)服務(wù)器可能存在瓶頸。（圖片來自網(wǎng)絡(luò)）

3.2.2 ring-reduce

百度提出的ring-reduce摒棄了參數(shù)服務(wù)器，采用環(huán)狀結(jié)構(gòu)來更新參數(shù)。ring-reduce把所有的worker組成一個(gè)兩兩相鄰的環(huán)形結(jié)構(gòu)。每個(gè)worker只與相鄰的worker交換參數(shù)。經(jīng)過幾次交換之后，所有的worker都包含其他worker的參數(shù)信息，達(dá)到更新的目的。（圖片來自網(wǎng)絡(luò)）

下面幾張圖，可以看到其中的幾個(gè)步驟；ring-reduce為了加快速度，并不是一次性交換所有的參數(shù)；而是先把參數(shù)進(jìn)行分割，不斷交換分割后參數(shù)。

4. 實(shí)現(xiàn)框架：Horovod

Horovod 是 Uber 開源的又一個(gè)深度學(xué)習(xí)工具，它的發(fā)展吸取了 Facebook「一小時(shí)訓(xùn)練 ImageNet 論文」與百度 Ring Allreduce 的優(yōu)點(diǎn)，可為用戶實(shí)現(xiàn)分布式訓(xùn)練提供幫助。https://github.com/horovod/horovod

采用NCCL 替換百度的 ring-allreduce 實(shí)現(xiàn)。NCCL 是英偉達(dá)的集合通信庫，提供高度優(yōu)化的 ring-allreduce 版本。NCCL 2 允許在多個(gè)機(jī)器之間運(yùn)行 ring-allreduc。

如果要把單機(jī)的訓(xùn)練代碼修改成分布式的代碼，只要幾個(gè)步驟就可以了 改造分布式訓(xùn)練：

• horovod安裝 建議安裝docker的horovod，省去安裝環(huán)境的麻煩。horovod依賴NCCL 2 open MPI

  $?mkdir?horovod-docker-gpu
  $?wget?-O?horovod-docker-gpu/Dockerfile?https://raw.githubusercontent.com/horovod/horovod/master/Dockerfile.gpu
  $?docker?build?-t?horovod:latest?horovod-docker-gpu
  

• 機(jī)器worker機(jī)器之間ssh打通

• 修改訓(xùn)練代碼 horovod支持tf,keras,pytorch和mxnet等不同的深度學(xué)習(xí)框架。以keras為例，修改主要6個(gè)步驟 （1） 初始化：hvd.init() （2）分配GPU計(jì)算資源：config.gpu_options.visible_device_list = str(hvd.local_rank())（3）分布式的優(yōu)化器來實(shí)現(xiàn)參數(shù)的分布式更新：opt = hvd.DistributedOptimizer(opt)（4）定義所有worker模型初始化一致性 hvd.callbacks.BroadcastGlobalVariablesCallback(0)（5）模型保存在某一個(gè)worker

  from?__future__?import?print_function
  import?keras
  from?keras.datasets?import?mnist
  from?keras.models?import?Sequential
  from?keras.layers?import?Dense,?Dropout,?Flatten
  from?keras.layers?import?Conv2D,?MaxPooling2D
  from?keras?import?backend?as?K
  import?math
  import?tensorflow?as?tf
  import?horovod.keras?as?hvd
  
  #?Horovod:?initialize?Horovod.
  hvd.init()
  
  #?Horovod:?pin?GPU?to?be?used?to?process?local?rank?(one?GPU?per?process)
  config?=?tf.ConfigProto()
  config.gpu_options.allow_growth?=?True
  config.gpu_options.visible_device_list?=?str(hvd.local_rank())
  K.set_session(tf.Session(config=config))
  
  batch_size?=?128
  num_classes?=?10
  
  #?Horovod:?adjust?number?of?epochs?based?on?number?of?GPUs.
  epochs?=?int(math.ceil(12.0?/?hvd.size()))
  
  #?Input?image?dimensions
  img_rows,?img_cols?=?28,?28
  
  #?The?data,?shuffled?and?split?between?train?and?test?sets
  (x_train,?y_train),?(x_test,?y_test)?=?mnist.load_data()
  
  if?K.image_data_format()?==?'channels_first':
  ????x_train?=?x_train.reshape(x_train.shape[0],?1,?img_rows,?img_cols)
  ????x_test?=?x_test.reshape(x_test.shape[0],?1,?img_rows,?img_cols)
  ????input_shape?=?(1,?img_rows,?img_cols)
  else:
  ????x_train?=?x_train.reshape(x_train.shape[0],?img_rows,?img_cols,?1)
  ????x_test?=?x_test.reshape(x_test.shape[0],?img_rows,?img_cols,?1)
  ????input_shape?=?(img_rows,?img_cols,?1)
  
  x_train?=?x_train.astype('float32')
  x_test?=?x_test.astype('float32')
  x_train?/=?255
  x_test?/=?255
  print('x_train?shape:',?x_train.shape)
  print(x_train.shape[0],?'train?samples')
  print(x_test.shape[0],?'test?samples')
  
  #?Convert?class?vectors?to?binary?class?matrices
  y_train?=?keras.utils.to_categorical(y_train,?num_classes)
  y_test?=?keras.utils.to_categorical(y_test,?num_classes)
  
  model?=?Sequential()
  model.add(Conv2D(32,?kernel_size=(3,?3),
  ????????????????activation='relu',
  ????????????????input_shape=input_shape))
  model.add(Conv2D(64,?(3,?3),?activation='relu'))
  model.add(MaxPooling2D(pool_size=(2,?2)))
  model.add(Dropout(0.25))
  model.add(Flatten())
  model.add(Dense(128,?activation='relu'))
  model.add(Dropout(0.5))
  model.add(Dense(num_classes,?activation='softmax'))
  
  #?Horovod:?adjust?learning?rate?based?on?number?of?GPUs.
  opt?=?keras.optimizers.Adadelta(1.0?*?hvd.size())
  
  #?Horovod:?add?Horovod?Distributed?Optimizer.
  opt?=?hvd.DistributedOptimizer(opt)
  
  model.compile(loss=keras.losses.categorical_crossentropy,
  ????????????optimizer=opt,
  ????????????metrics=['accuracy'])
  
  callbacks?=?[
  ????#?Horovod:?broadcast?initial?variable?states?from?rank?0?to?all?other?processes.
  ????#?This?is?necessary?to?ensure?consistent?initialization?of?all?workers?when
  ????#?training?is?started?with?random?weights?or?restored?from?a?checkpoint.
  ????hvd.callbacks.BroadcastGlobalVariablesCallback(0),
  ]
  
  #?Horovod:?save?checkpoints?only?on?worker?0?to?prevent?other?workers?from?corrupting?them.
  if?hvd.rank()?==?0:
  ????callbacks.append(keras.callbacks.ModelCheckpoint('./checkpoint-{epoch}.h5'))
  
  model.fit(x_train,?y_train,
  ????????batch_size=batch_size,
  ????????callbacks=callbacks,
  ????????epochs=epochs,
  ????????verbose=1,
  ????????validation_data=(x_test,?y_test))
  score?=?model.evaluate(x_test,?y_test,?verbose=0)
  print('Test?loss:',?score[0])
  print('Test?accuracy:',?score[1])
  

• 利用horovodrun 執(zhí)行分布式訓(xùn)練

horovodrun?-np?16?-H?server1:4,server2:4,server3:4,server4:4?python?train.py

5. 總結(jié)

本文分享了通過GPU利用率和分布式訓(xùn)練Horovod框架來提升深度學(xué)習(xí)訓(xùn)練。

• 并行CPU加載數(shù)據(jù)和預(yù)處理，讓GPU不再等待CPU
• 采用Horovod讓數(shù)據(jù)并行來提高大數(shù)據(jù)量的訓(xùn)練的迭代時(shí)間