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

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

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

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

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

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

在實(shí)踐中，我們將訓(xùn)練數(shù)據(jù)的微型批次分為幾個(gè)部分，例如4、8、16等（在本文中，我們將使用術(shù)語“子批次”來指代這些拆分的部分），并且每個(gè)培訓(xùn)工人分一個(gè)批次。然后，培訓(xùn)人員分別使用子批進(jìn)行前饋、梯度計(jì)算和模型更新，就像在整體培訓(xùn)模式中一樣。在這些步驟之后，將調(diào)用稱為模型平均的過程，對參與培訓(xùn)的所有工作人員的模型參數(shù)求平均，以便在新的培訓(xùn)迭代開始時(shí)使模型參數(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í)，我們一直等到模型參數(shù)達(dá)到一致性為止，因此可以在新的迭代開始之前使模型同步。這種方法稱為同步SGD，這是我們將在本文中考慮的方法。（另一種方法是異步SGD，在異步SGD中，模型參數(shù)存儲在稱為參數(shù)服務(wù)器的集中位置，并且工作線程在每次迭代結(jié)束時(shí)自行獨(dú)立于參數(shù)服務(wù)器進(jìn)行更新，而與其他工作線程的狀態(tài)無關(guān)。同步SGD、異步SGD的整體訓(xùn)練速度不會(huì)受到單個(gè)“慢”工人的影響，但是，如果訓(xùn)練集群中的GPU處理速度大致相同（通常是這種情況），在實(shí)際情況中顯然很“慢”，因此，同步SGD在ML應(yīng)用領(lǐng)域中是一個(gè)不錯(cuò)的選擇。）

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

“參數(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)。在這種方法中，計(jì)算節(jié)點(diǎn)被劃分為工作程序和參數(shù)服務(wù)器。每個(gè)工作人員“擁有”一部分?jǐn)?shù)據(jù)和工作負(fù)載，并且參數(shù)服務(wù)器共同維護(hù)全局共享的參數(shù)（使用參數(shù)服務(wù)器擴(kuò)展分布式機(jī)器學(xué)習(xí)）。在每次迭代的開始，工作人員會(huì)提取完整的模型參數(shù)副本，并在迭代結(jié)束時(shí)將新更新的模型推回參數(shù)服務(wù)器。對于同步SGD，參數(shù)服務(wù)器將平均所有工作人員推送的模型參數(shù)，從而創(chuàng)建更新的“全局”模型供工作人員在下一次迭代開始時(shí)提取。

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

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

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

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

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

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

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

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

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

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

await_workers函數(shù)用于等待其他由其會(huì)話ID指定的CML容器的啟動(dòng)。 await_workers的返回值是一個(gè)Python字典，帶有一個(gè)項(xiàng)的鍵名是ip_address，并帶有其IP地址。下面的代碼顯示了如何在CML主會(huì)話中使用cdsw_await_workers。請注意，如果某些容器在指定的時(shí)間（例如，以下代碼為60秒）后無法啟動(dòng)，則await_workers的返回值將導(dǎo)致鍵名失敗的項(xiàng)，其中包含失敗人員的會(huì)話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è)工作程序都需要構(gòu)造實(shí)例。在下面的代碼段中，PS1:PORT1代表第一個(gè)TF參數(shù)服務(wù)器進(jìn)程的IP地址和端口號，PS2:PORT2代表第二個(gè)TF參數(shù)服務(wù)器進(jìn)程的IP地址和端口號，而WORKER1:PORT1代表第二個(gè)TF參數(shù)服務(wù)器進(jìn)程的IP地址和端口號。 第一個(gè)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ù)器進(jìn)程和輔助進(jìn)程都加入集群。

server.join()

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

• 使用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進(jìn)行培訓(xùn)，因?yàn)镸onitoredTrainingSession提供了許多監(jiān)視和自動(dòng)管理功能，這些功能對于高效的分布式模型培訓(xùn)環(huán)境至關(guān)重要。

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

每次我們編寫分布式Tensorflow代碼時(shí)，重復(fù)上述上述編程過程不僅很耗時(shí)，而且容易出錯(cuò)。因此，我們將它們包裝在一個(gè)函數(shù)（cdsw_tensorflow_utils.run_cluster）中，該函數(shù)隨此文章一起發(fā)布，從而使整個(gè)過程自動(dòng)化，因此數(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進(jìn)行了實(shí)現(xiàn)。

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

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

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

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

Host 100.66.0.29
Port 2222
Host 100.66.0.30
Port 2222

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

接下來，啟動(dòng)幾個(gè)CML工作器容器，并等待直到獲得啟動(dòng)的工作器的IP地址。 （此方法與基于參數(shù)服務(wù)器的分布式深度學(xué)習(xí)方法中使用的方法完全相同。）然后，在這些啟動(dòng)的工作程序中啟動(dòng)Horovod模型訓(xùn)練過程，可以通過在os.system中調(diào)用horovodrun命令來實(shí)現(xiàn)() Python函數(shù)。這兩個(gè)步驟都可以通過CML主會(huì)話中的Python代碼完成。以下是實(shí)現(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個(gè)主要修改：1）在工作進(jìn)程之間創(chuàng)建集群環(huán)境，以及2）執(zhí)行模型平均。除了對MPI Allreduce方法及其在CML中的實(shí)現(xiàn)進(jì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推動(dòng)了全球最大型企業(yè)的數(shù)字化轉(zhuǎn)型歷程。了解更多，請聯(lián)系。

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

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