許多企業(yè)數(shù)據(jù)科學(xué)團隊正在使用Cloudera的機器學(xué)習(xí)平臺進行模型探索和培訓(xùn)，包括使用Tensorflow，PyTorch等創(chuàng)建深度學(xué)習(xí)模型。但是，訓(xùn)練深度學(xué)習(xí)模型通常是一個耗時的過程，因此采用GPU和分布式模型訓(xùn)練方法來加快訓(xùn)練速度。

這是我們關(guān)于在Cloudera機器學(xué)習(xí)平臺上進行深度學(xué)習(xí)的分布式模型訓(xùn)練的博客系列的第一篇文章，其中包括Cloudera數(shù)據(jù)科學(xué)工作臺（CDSW）和Cloudera機器學(xué)習(xí)（CML），這是為云構(gòu)建的新一代CDSW。在下文中，為簡單起見，我們僅指“CML”，且此文的內(nèi)容也適用于CDSW安裝。

在這篇文章中，我們將介紹：

• 深度學(xué)習(xí)的分布式模型訓(xùn)練的基本方法
• 如何將這些方法集成到CML中的方法
• 新功能可自動執(zhí)行基于參數(shù)服務(wù)器的分布式Tensorflow模型訓(xùn)練
• 以及使用即將推出的CML功能啟用MPI Allreduce分布式模型訓(xùn)練的預(yù)覽。

深度學(xué)習(xí)的分布式模型訓(xùn)練的技術(shù)基礎(chǔ)

通常使用隨機梯度后裔（SGD）算法訓(xùn)練深度學(xué)習(xí)模型。對于SGD的每次迭代，我們將從訓(xùn)練集中采樣一個小批量，將其輸入到訓(xùn)練模型中，計算觀察值和實際值的損失函數(shù)的梯度，并更新模型參數(shù)（或權(quán)重）。眾所周知，SGD迭代必須順序執(zhí)行，因此不可能通過并行化迭代來加快訓(xùn)練過程。但是，由于使用CIFAR10或IMAGENET等許多常用模型處理一次迭代要花費很長時間，即使使用最先進的GPU，我們?nèi)匀豢梢試L試并行化前饋計算以及每次迭代中的梯度計算以加快速度加快模型訓(xùn)練過程。

在實踐中，我們將訓(xùn)練數(shù)據(jù)的微型批次分為幾個部分，例如4、8、16等（在本文中，我們將使用術(shù)語“子批次”來指代這些拆分的部分），并且每個培訓(xùn)工人分一個批次。然后，培訓(xùn)人員分別使用子批進行前饋、梯度計算和模型更新，就像在整體培訓(xùn)模式中一樣。在這些步驟之后，將調(diào)用稱為模型平均的過程，對參與培訓(xùn)的所有工作人員的模型參數(shù)求平均，以便在新的培訓(xùn)迭代開始時使模型參數(shù)完全相同。然后，新一輪的訓(xùn)練迭代又從數(shù)據(jù)采樣和拆分步驟開始。

形式上，上面的分布式模型訓(xùn)練過程的模型平均的一般思想可以使用以下偽代碼表示。

# Suppose w0 is the initial global parameters, K is the number of workers, T is the overall iterating number, ftk(wt) is the output of the kth worker under the parameters wt at time t, and lr is the learning rate.
FOR t = 0, 1, …, T-1 DO
    Read the current model parameters wt
    Stochastically sample a batch of data itk
    Compute the stochastic gradients ??ftk(wt) at each worker
    Accumulate all of the gradients of K workers
    Update the global parameters wt+1=wt-ltK??ftk(wt)
END FOR

如上所示，在每次迭代結(jié)束時，我們一直等到模型參數(shù)達到一致性為止，因此可以在新的迭代開始之前使模型同步。這種方法稱為同步SGD，這是我們將在本文中考慮的方法。（另一種方法是異步SGD，在異步SGD中，模型參數(shù)存儲在稱為參數(shù)服務(wù)器的集中位置，并且工作線程在每次迭代結(jié)束時自行獨立于參數(shù)服務(wù)器進行更新，而與其他工作線程的狀態(tài)無關(guān)。同步SGD、異步SGD的整體訓(xùn)練速度不會受到單個“慢”工人的影響，但是，如果訓(xùn)練集群中的GPU處理速度大致相同（通常是這種情況），在實際情況中顯然很“慢”，因此，同步SGD在ML應(yīng)用領(lǐng)域中是一個不錯的選擇。）

上述算法中的每個工作人員在訓(xùn)練過程中還具有模型參數(shù)的完整副本，只有訓(xùn)練數(shù)據(jù)分配給不同的工作人員。這種方法稱為數(shù)據(jù)并行性，這是我們將在此處考慮的方法。另一種稱為模型并行性的方法也可以拆分模型參數(shù)。模型并行性的優(yōu)點是能夠訓(xùn)練大于內(nèi)存容量（主內(nèi)存或GPU內(nèi)存）的模型。但是，如果模型的大小小于內(nèi)存容量，則數(shù)據(jù)并行性將更加高效，因為在每次迭代的前饋期間它不需要工作人員之間的通信。

“參數(shù)服務(wù)器”對比“MPI Allreduce”

許多深度學(xué)習(xí)框架，例如Tensorflow，PyTorch和Horovod，都支持分布式模型訓(xùn)練。它們在模型參數(shù)的平均或同步方式上有很大不同。當(dāng)前，有兩種模型同步方法：1）基于參數(shù)服務(wù)器，和2）MPI Allreduce。

上圖顯示了基于參數(shù)服務(wù)器的體系結(jié)構(gòu)。在這種方法中，計算節(jié)點被劃分為工作程序和參數(shù)服務(wù)器。每個工作人員“擁有”一部分數(shù)據(jù)和工作負載，并且參數(shù)服務(wù)器共同維護全局共享的參數(shù)（使用參數(shù)服務(wù)器擴展分布式機器學(xué)習(xí)）。在每次迭代的開始，工作人員會提取完整的模型參數(shù)副本，并在迭代結(jié)束時將新更新的模型推回參數(shù)服務(wù)器。對于同步SGD，參數(shù)服務(wù)器將平均所有工作人員推送的模型參數(shù)，從而創(chuàng)建更新的“全局”模型供工作人員在下一次迭代開始時提取。

另一方面，MPI Allreduce方法不需要一組專用服務(wù)器來存儲參數(shù)。取而代之的是，它利用環(huán)減少（將HPC技術(shù)帶入深度學(xué)習(xí)）算法和MPI（消息傳遞接口）API來實現(xiàn)模型同步。對于由N個節(jié)點組成的模型訓(xùn)練集群，模型參數(shù)將被劃分為N個塊，并且參與環(huán)歸約算法的每個節(jié)點都將與其兩個對等節(jié)點進行2?(N?1)次通信。因此，從理論上講，模型平均時間僅與模型的大小有關(guān)，而與節(jié)點的數(shù)量無關(guān)。在此通信期間，節(jié)點發(fā)送和接收數(shù)據(jù)緩沖區(qū)的塊。在前N-1次迭代中，將接收到的值添加到節(jié)點緩沖區(qū)中的值。在第二次N-1迭代中，接收到的值替換了保存在節(jié)點緩沖區(qū)中的值。MPI API是由高性能計算社區(qū)開發(fā)的，用于實現(xiàn)模型參數(shù)同步，而Open MPI是由學(xué)術(shù)，研究和行業(yè)合作伙伴組成的聯(lián)盟開發(fā)和維護的，廣泛使用的MPI實現(xiàn)之一。

關(guān)于與MPI Allreduce方法相比基于參數(shù)服務(wù)器的方法的性能，在Uber和MXNet中的基準(zhǔn)測試結(jié)果表明，在小數(shù)量的節(jié)點（8-64）上，MPI Allreduce的性能優(yōu)于參數(shù)服務(wù)器（Horovod：快速而輕松在TensorFlow中進行分布式深度學(xué)習(xí)，并通過MPI AllReduce擴展MXNet分布式培訓(xùn)）。

上圖是Uber的基準(zhǔn)測試，參數(shù)服務(wù)器（Tensorflow本機）與MPI Allreduce（Horovod）的結(jié)果，當(dāng)在不同數(shù)量的NVIDIA Pascal GPU上運行分布式培訓(xùn)作業(yè)時，將每秒處理的圖像與標(biāo)準(zhǔn)分布式TensorFlow和Horovod進行比較適用于基于25GbE TCP的Inception V3和ResNet-101 TensorFlow模型。同時，下面的MXNet基準(zhǔn)測試結(jié)果還顯示，即使參數(shù)服務(wù)器和輔助服務(wù)器的數(shù)量均為8，MPI Allreduce方法的性能仍高于參數(shù)服務(wù)器方法。

從性能數(shù)據(jù)中，我們可以得出以下結(jié)論（由MPI AllReduce擴展MXNet分布式培訓(xùn)）：1）MPI Allreduce方法不需要額外的服務(wù)器節(jié)點，并且可以獲得比基于參數(shù)服務(wù)器的方法更好的性能。同步SGD多節(jié)點訓(xùn)練。（在基于參數(shù)服務(wù)器的方法中，如果配置不當(dāng)，則不足的服務(wù)器將成為網(wǎng)絡(luò)帶寬的熱點。）2）此外，MPI Allreduce方法更易于硬件部署。（在基于參數(shù)服務(wù)器的方法中，需要精心計算服務(wù)器：工人比率的配置，并且該比率不是固定的（取決于拓撲和網(wǎng)絡(luò)帶寬）。）

傳統(tǒng)上，Tensorflow支持基于參數(shù)服務(wù)器的方法，而PyTorch和Horovod支持MPI Allreduce方法。但是，從r1.3開始，Tensorflow也開始支持MPI Allreduce方法（在r1.4中具有實驗支持）。

注意：基于參數(shù)服務(wù)器的方法能夠支持同步和異步SGD，例如Tensorflow。據(jù)我們所知，MPI Allreduce方法的所有當(dāng)前實現(xiàn)僅支持同步SGD。

有了這些基礎(chǔ)知識，讓我們繼續(xù)進行分布式模型訓(xùn)練的編程部分，同時使用基于參數(shù)服務(wù)器的參數(shù)和MPI Allreduce方法，并了解如何在CML中使用這兩種方法。

使用CML中基于參數(shù)服務(wù)器的方法進行編程

本節(jié)將概述用CML編寫基于參數(shù)服務(wù)器的分布式模型訓(xùn)練代碼的概述。我們將使用Tensorflow本機分布式API和CML的分布式API（cdsw.launch_worker）進行演示。

首先，分布式Tensorflow中的每個參數(shù)服務(wù)器或工作程序都是一個Python進程。因此，我們很自然地使用CML工作器（或容器）來表示TF參數(shù)服務(wù)器或TF工作器，并使用cdsw.launch_workers(…)函數(shù)在主CML會話中調(diào)用這些CML工作器。在cdsw.launch_workers(…)中，我們還可以為TF參數(shù)服務(wù)器和TF工作者指定不同的Python程序文件。然后，主要的CML會話需要收集每個容器的主機名或IP地址，并將它們發(fā)送給所有CML子工作程序，以創(chuàng)建集群規(guī)范（tf.train.ClusterSpec）。在CML中，實際上有許多方法可以獲取每個子工作者的IP地址，我們將介紹一種使用新的await_workers函數(shù)的方法，該函數(shù)在CML Docker引擎V10中正式可用。

await_workers函數(shù)用于等待其他由其會話ID指定的CML容器的啟動。 await_workers的返回值是一個Python字典，帶有一個項的鍵名是ip_address，并帶有其IP地址。下面的代碼顯示了如何在CML主會話中使用cdsw_await_workers。請注意，如果某些容器在指定的時間（例如，以下代碼為60秒）后無法啟動，則await_workers的返回值將導(dǎo)致鍵名失敗的項，其中包含失敗人員的會話ID。

# CML main session
import cdsw
workers = cdsw.launch_workers(NUM_WORKERS,
                              cpu=0.5, memory=2,
                              script=”...”)
worker_ids = [worker["id"] for worker in workers]
running_workers = cdsw.await_workers(worker_ids,
                              wait_for_completion=False,
                              timeout_seconds=60)
worker_ips = [worker["ip_address"] for worker in \
                              running_workers["workers"]]

在獲取并分配所有TF參數(shù)服務(wù)器和TF工作程序的IP地址之后，每個工作程序都需要構(gòu)造實例。在下面的代碼段中，PS1:PORT1代表第一個TF參數(shù)服務(wù)器進程的IP地址和端口號，PS2:PORT2代表第二個TF參數(shù)服務(wù)器進程的IP地址和端口號，而WORKER1:PORT1代表第二個TF參數(shù)服務(wù)器進程的IP地址和端口號。 第一個TF工作程序的IP地址和端口號等

cluster = tf.train.ClusterSpec({"ps": ["PS1:PORT1","PS2:PORT2",...],
                     "worker": ["WORKER1:PORT1","WORKER2:PORT2",...]})
server = tf.train.Server(cluster, job_name="PS or WORKER",
                                  task_index=NUM)

對于TF參數(shù)服務(wù)器容器，請調(diào)用server.join()等待，直到所有其他參數(shù)服務(wù)器進程和輔助進程都加入集群。

server.join()

對于TF工作人員，所有的建模和培訓(xùn)代碼都需要進行編程。如果您使用數(shù)據(jù)并行性，那么建模部分實際上與整體式Tensorflow程序相同。但是，該訓(xùn)練代碼與整體Tensorflow程序至少有2個明顯的不同。

• 使用tf.train.SyncReplicasOptimizer包裝模型優(yōu)化器，tf.train.SyncReplicasOptimizer是整體Tensorflow優(yōu)化器的同步SGD版本。這是tf.train.SyncReplicasOptimizer的示例代碼。

optimizer = tf.train.AdamOptimizer(learning_rate=...)
sr_optim = tf.train.SyncReplicasOptimizer(
                     optimizer,
                     replicas_to_aggregate=NUM_WORKER,
                     total_num_replicas=NUM_WORKER)

• 通常，建議使用tf.train.MonitoredTrainingSession代替tf.Session進行培訓(xùn)，因為MonitoredTrainingSession提供了許多監(jiān)視和自動管理功能，這些功能對于高效的分布式模型培訓(xùn)環(huán)境至關(guān)重要。

注意：在最新版本的Tensorflow中不推薦使用tf.train.Supervisor，現(xiàn)在建議使用tf.train.MonitoredTrainingSession代替tf.train.Supervisor。

每次我們編寫分布式Tensorflow代碼時，重復(fù)上述上述編程過程不僅很耗時，而且容易出錯。因此，我們將它們包裝在一個函數(shù)（cdsw_tensorflow_utils.run_cluster）中，該函數(shù)隨此文章一起發(fā)布，從而使整個過程自動化，因此數(shù)據(jù)科學(xué)家僅需指定參數(shù)服務(wù)器，工作程序和培訓(xùn)代碼的數(shù)量即可。可以在此處找到包含該功能的腳本。以下程序演示了如何使用cdsw_tensorflow_utils.run_cluster創(chuàng)建分布式Tensorflow集群。

cluster_spec, session_addr = cdsw_tensorflow_utils.run_cluster(
                                 n_workers=n_workers,
                                 n_ps=n_ps,
                                 cpu=0.5,
                                 memory=2,
                                 worker_script="train.py")

文件train.py是模型定義和訓(xùn)練代碼所在的地方，它看起來很像單片Tensorflow代碼。train.py的結(jié)構(gòu)如下：

import sys, time
import tensorflow as tf

 
# config model training parameters
batch_size = 100
learning_rate = 0.0005
training_epochs = 20

 
# load data set
from tensorflow.examples.tutorials.mnist import input_data
mnist = input_data.read_data_sets('MNIST_data', one_hot=True)

 
# Define the run() function with the following arguments
# And this function will be invoked within CML API
def run(cluster, server, task_index):

 
    # Specify cluster and device in tf.device() function
    with tf.device(tf.train.replica_device_setter(
        worker_device="/job:worker/task:%d" % task_index,
        cluster=cluster)):

 
        # Count the number of updates
        global_step = tf.get_variable(
            'global_step',
            [],
            initializer = tf.constant_initializer(0),
            trainable = False)


        # Model definition
        …

 
    # Define a tf.train.Supervisor instance
    # and use it to start model training
    sv = tf.train.Supervisor(is_chief=(task_index == 0),
                            global_step=global_step,
                            init_op=init_op)
    with sv.prepare_or_wait_for_session(server.target) as sess:

 
        # Model training code
        …

 
    # Stop the Supervisor instance
    sv.stop()

下面的屏幕快照顯示了使用上面介紹的CML內(nèi)置API在CML上平均異步模型的分布式模型訓(xùn)練程序的執(zhí)行過程。

MPI Allreduce CML編程方法預(yù)覽

本節(jié)概述了使用CML中的MPI Allreduce方法執(zhí)行分布式模型訓(xùn)練的過程，并使用Horovod進行了實現(xiàn)。

使用Horovod時，驅(qū)動程序節(jié)點（在此上下文中為CML主會話）需要執(zhí)行SSH無密碼登錄到Horovod輔助節(jié)點，以啟動所有模型訓(xùn)練過程。 要在CML中為用戶cdsw啟用SSH無密碼登錄，需要兩個設(shè)置步驟：1）為用戶cdsw設(shè)置無密碼身份驗證，以及2）指定默認的SSH偵聽端口從22到2222。

• 在CML中為用戶cdsw啟用SSH無密碼身份驗證非常容易。您只需要轉(zhuǎn)到用戶設(shè)置頁面，然后將公鑰從“outbound ssh”標(biāo)簽復(fù)制到“remote editing”標(biāo)簽即可。在那之后，同一用戶內(nèi)的所有會話都可以使用無密碼登錄相互進行SSH交換。

注意：CML的April 14及更高版本不需要此步驟。

• CML引擎的sshd偵聽端口為2222。使用mpirun命令調(diào)用Horovod worker時，可以創(chuàng)建~/.ssh/config文件，并輸入如下內(nèi)容。

Host 100.66.0.29
Port 2222
Host 100.66.0.30
Port 2222

否則，使用horovodrun命令時，只需為horovodrun指定一個額外的參數(shù)-p 2222。

接下來，啟動幾個CML工作器容器，并等待直到獲得啟動的工作器的IP地址。 （此方法與基于參數(shù)服務(wù)器的分布式深度學(xué)習(xí)方法中使用的方法完全相同。）然后，在這些啟動的工作程序中啟動Horovod模型訓(xùn)練過程，可以通過在os.system中調(diào)用horovodrun命令來實現(xiàn)() Python函數(shù)。這兩個步驟都可以通過CML主會話中的Python代碼完成。以下是實現(xiàn)功能的示例代碼，train.py只是用于模型訓(xùn)練的Python代碼。

# CML main session
Import os
import cdsw

 
workers = cdsw.launch_workers(NUM_WORKERS,
                              cpu=0.5, memory=2,
                              script=”...”)
worker_ids = [worker["id"] for worker in workers]
running_workers = cdsw.await_workers(worker_ids,
                              wait_for_completion=False,
                              timeout_seconds=60)
worker_ips = [worker["ip_address"] for worker in \
                              Running_workers["workers"]]
cmd="horovodrun -np {} -H {} -p 2222 python train.py".format(
                              len(worker_ips),
                              ",".join(worker_ips))

 
os.system(cmd)

在MPI Allreduce方法中，仍然需要修改模型訓(xùn)練文件，即上面的示例代碼的train.py。在train.py中，代碼有2個主要修改：1）在工作進程之間創(chuàng)建集群環(huán)境，以及2）執(zhí)行模型平均。除了對MPI Allreduce方法及其在CML中的實現(xiàn)進行技術(shù)深入研究之外，我們還將在本系列的下一篇文章中討論這些主題，并介紹這些方法的性能基準(zhǔn)測試結(jié)果。

關(guān)于Cloudera

在 Cloudera，我們相信數(shù)據(jù)可以使今天的不可能，在明天成為可能。我們使人們能夠?qū)?fù)雜的數(shù)據(jù)轉(zhuǎn)換為清晰而可行的洞察力。Cloudera 為任何地方的任何數(shù)據(jù)從邊緣到人工智能提供企業(yè)數(shù)據(jù)云平臺服務(wù)。在開源社區(qū)不懈創(chuàng)新的支持下， Cloudera推動了全球最大型企業(yè)的數(shù)字化轉(zhuǎn)型歷程。了解更多，請聯(lián)系。

慧都大數(shù)據(jù)專業(yè)團隊為企業(yè)提供Cloudera大數(shù)據(jù)平臺搭建，免費業(yè)務(wù)咨詢，定制開發(fā)等完整服務(wù)，快速、輕松、低成本將任何Hadoop集群從試用階段轉(zhuǎn)移到生產(chǎn)階段。

歡迎撥打慧都熱線023-68661681或咨詢慧都在線客服，我們有專業(yè)的大數(shù)據(jù)團隊，為您提供免費大數(shù)據(jù)相關(guān)業(yè)務(wù)咨詢！