調(diào)參俠看過(guò)來(lái)！兩個(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ù)大，在百萬(wàn)/千萬(wàn)的量級(jí)，訓(xùn)練一個(gè)Epoch需要很長(zhǎng)時(shí)間，模型迭代周期過(guò)長(zhǎng)。
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)練（圖片來(lái)自網(wǎng)絡(luò)）
2.2 解決方案
采用多進(jìn)程并行處理，加快CPU加載數(shù)據(jù)的性能
- keras keras 中提供了workers use_multiprocessing來(lái)采用多進(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},
- epochsepochs=epochs,
- verbose=1,
- steps_per_epochsteps_per_epoch=steps_per_epoch,
- callbacks=callbacks_list,
- validation_data=valid_generator,
- validation_stepsvalidation_steps=validation_steps,
- shuffle=True,
- workers=8,
- use_multiprocessing=True,
- max_queue_size=20
- pytorch torch在加載數(shù)據(jù)中提供類(lèi)似參數(shù)num_workers。pin_memory=True可以直接加載到顯存中，而不需要內(nèi)存
- torch.utils.data.DataLoader(image_datasets[x],
- batch_sizebatch_size=batch_size,
- shuffle=True,
- num_workers=8,
- pin_memory=True)
3. 分布式并行訓(xùn)練
3.1 并行模式
當(dāng)訓(xùn)練的數(shù)據(jù)量很大時(shí)，可以通過(guò)多個(gè)機(jī)器多個(gè)GPU來(lái)提高訓(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)系，不能完全的并行。（圖片來(lái)自網(wǎng)絡(luò)）
- 數(shù)據(jù)并行（ data parallelism ）：不同的機(jī)器有同一個(gè)模型的多個(gè)副本，每個(gè)機(jī)器分配到不同的數(shù)據(jù)，然后將所有機(jī)器的計(jì)算結(jié)果按照某種方式合并。這種就比較適合大數(shù)據(jù)的情況。數(shù)據(jù)并行要解決的問(wèn)題是數(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ú)服務(wù)器模式）
3.2.1 參數(shù)服務(wù)器模式
參數(shù)服務(wù)器模式，見(jiàn)下圖。在每個(gè)worker執(zhí)行完一個(gè)batch的訓(xùn)練后，反向傳播參數(shù)的時(shí)候，所有的worker都會(huì)把參數(shù)傳給參數(shù)服務(wù)器，進(jìn)行匯總求均值，之后再傳給每個(gè)worker，進(jìn)入第二個(gè)batch的訓(xùn)練。（圖片來(lái)自網(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ù)器可能存在瓶頸。（圖片來(lái)自網(wǎng)絡(luò)）
3.2.2 ring-reduce
百度提出的ring-reduce摒棄了參數(shù)服務(wù)器，采用環(huán)狀結(jié)構(gòu)來(lái)更新參數(shù)。ring-reduce把所有的worker組成一個(gè)兩兩相鄰的環(huán)形結(jié)構(gòu)。每個(gè)worker只與相鄰的worker交換參數(shù)。經(jīng)過(guò)幾次交換之后，所有的worker都包含其他worker的參數(shù)信息，達(dá)到更新的目的。（圖片來(lái)自網(wǎng)絡(luò)）
下面幾張圖，可以看到其中的幾個(gè)步驟；ring-reduce為了加快速度，并不是一次性交換所有的參數(shù)；而是先把參數(shù)進(jìn)行分割，不斷交換分割后參數(shù)。
4. 實(shí)現(xiàn)框架：Horovod
Horovod 是 Uber 開(kāi)源的又一個(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á)的集合通信庫(kù)，提供高度優(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)化器來(lái)實(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(configconfig=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_trainx_train = x_train.reshape(x_train.shape[0], 1, img_rows, img_cols)
- x_testx_test = x_test.reshape(x_test.shape[0], 1, img_rows, img_cols)
- input_shape = (1, img_rows, img_cols)
- else:
- x_trainx_train = x_train.reshape(x_train.shape[0], img_rows, img_cols, 1)
- x_testx_test = x_test.reshape(x_test.shape[0], img_rows, img_cols, 1)
- input_shape = (img_rows, img_cols, 1)
- x_trainx_train = x_train.astype('float32')
- x_testx_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_shapeinput_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,
- optoptimizer=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_sizebatch_size=batch_size,
- callbackscallbacks=callbacks,
- epochsepochs=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é)
本文分享了通過(guò)GPU利用率和分布式訓(xùn)練Horovod框架來(lái)提升深度學(xué)習(xí)訓(xùn)練。
- 并行CPU加載數(shù)據(jù)和預(yù)處理，讓GPU不再等待CPU
- 采用Horovod讓數(shù)據(jù)并行來(lái)提高大數(shù)據(jù)量的訓(xùn)練的迭代時(shí)間
