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

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

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

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

深度學習的分布式模型訓練的技術基礎

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

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

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

# 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

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

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

“參數服務器”對比“MPI Allreduce”

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

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

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

關于與MPI Allreduce方法相比基于參數服務器的方法的性能，在Uber和MXNet中的基準測試結果表明，在小數量的節點（8-64）上，MPI Allreduce的性能優于參數服務器（Horovod：快速而輕松在TensorFlow中進行分布式深度學習，并通過MPI AllReduce擴展MXNet分布式培訓）。

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

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

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

注意：基于參數服務器的方法能夠支持同步和異步SGD，例如Tensorflow。據我們所知，MPI Allreduce方法的所有當前實現僅支持同步SGD。

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

使用CML中基于參數服務器的方法進行編程

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

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

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

server.join()

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

• 使用tf.train.SyncReplicasOptimizer包裝模型優化器，tf.train.SyncReplicasOptimizer是整體Tensorflow優化器的同步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進行培訓，因為MonitoredTrainingSession提供了許多監視和自動管理功能，這些功能對于高效的分布式模型培訓環境至關重要。

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

每次我們編寫分布式Tensorflow代碼時，重復上述上述編程過程不僅很耗時，而且容易出錯。因此，我們將它們包裝在一個函數（cdsw_tensorflow_utils.run_cluster）中，該函數隨此文章一起發布，從而使整個過程自動化，因此數據科學家僅需指定參數服務器，工作程序和培訓代碼的數量即可。可以在此處找到包含該功能的腳本。以下程序演示了如何使用cdsw_tensorflow_utils.run_cluster創建分布式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是模型定義和訓練代碼所在的地方，它看起來很像單片Tensorflow代碼。train.py的結構如下：

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內置API在CML上平均異步模型的分布式模型訓練程序的執行過程。

MPI Allreduce CML編程方法預覽

本節概述了使用CML中的MPI Allreduce方法執行分布式模型訓練的過程，并使用Horovod進行了實現。

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

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

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

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

Host 100.66.0.29
Port 2222
Host 100.66.0.30
Port 2222

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

接下來，啟動幾個CML工作器容器，并等待直到獲得啟動的工作器的IP地址。 （此方法與基于參數服務器的分布式深度學習方法中使用的方法完全相同。）然后，在這些啟動的工作程序中啟動Horovod模型訓練過程，可以通過在os.system中調用horovodrun命令來實現() Python函數。這兩個步驟都可以通過CML主會話中的Python代碼完成。以下是實現功能的示例代碼，train.py只是用于模型訓練的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方法中，仍然需要修改模型訓練文件，即上面的示例代碼的train.py。在train.py中，代碼有2個主要修改：1）在工作進程之間創建集群環境，以及2）執行模型平均。除了對MPI Allreduce方法及其在CML中的實現進行技術深入研究之外，我們還將在本系列的下一篇文章中討論這些主題，并介紹這些方法的性能基準測試結果。

關于Cloudera

在 Cloudera，我們相信數據可以使今天的不可能，在明天成為可能。我們使人們能夠將復雜的數據轉換為清晰而可行的洞察力。Cloudera 為任何地方的任何數據從邊緣到人工智能提供企業數據云平臺服務。在開源社區不懈創新的支持下， Cloudera推動了全球最大型企業的數字化轉型歷程。了解更多，請聯系。

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

歡迎撥打慧都熱線023-68661681或咨詢慧都在線客服，我們有專業的大數據團隊，為您提供免費大數據相關業務咨詢！