[深度學(xué)習(xí)] 分布式模式介紹（一）

[深度學(xué)習(xí)] 分布式Tensorflow 2.0介紹（二）

[深度學(xué)習(xí)] 分布式Pytorch 1.0介紹（三）

[深度學(xué)習(xí)] 分布式Horovod介紹（四）

一 單GPU訓(xùn)練 vs 多GPU訓(xùn)練

單GPU訓(xùn)練 一般代碼比較簡(jiǎn)單，并且能滿足我們的基本需求，通常做法是設(shè)定變量CUDA_VISIBLE_DEVICES的值為某一塊GPU來(lái)Mask我們機(jī)器上的GPU設(shè)備，雖然有時(shí)當(dāng)我們忘了設(shè)定該變量時(shí)程序會(huì)自動(dòng)占用所有的GPU資源，但如果沒(méi)有相應(yīng)的代碼去分配掌控GPU資源的使用的話，程序還是只會(huì)利用到第一張卡的計(jì)算資源，其他的資源則僅是占用浪費(fèi)狀態(tài)。

多GPU訓(xùn)練 則可以從兩個(gè)方面提升我們模型訓(xùn)練的上限：1. 超過(guò)單卡顯存上限的模型大小， 2. 更大的Batch Size和更快訓(xùn)練速度。

單機(jī)的多GPU訓(xùn)練， tensorflow的官方已經(jīng)給了一個(gè)cifar的例子，已經(jīng)有比較詳細(xì)的代碼和文檔介紹， 這里大致說(shuō)下多GPU的過(guò)程，以便方便引入到多機(jī)多GPU的介紹。
單機(jī)多GPU的訓(xùn)練過(guò)程：

假設(shè)你的機(jī)器上有3個(gè)GPU;

在單機(jī)單GPU的訓(xùn)練中，數(shù)據(jù)是一個(gè)batch一個(gè)batch的訓(xùn)練。 在單機(jī)多GPU中，數(shù)據(jù)一次處理3個(gè)batch(假設(shè)是3個(gè)GPU訓(xùn)練）， 每個(gè)GPU處理一個(gè)batch的數(shù)據(jù)計(jì)算。

變量，或者說(shuō)參數(shù)，保存在CPU上

剛開(kāi)始的時(shí)候數(shù)據(jù)由CPU分發(fā)給3個(gè)GPU， 在GPU上完成了計(jì)算，得到每個(gè)batch要更新的梯度。

然后在CPU上收集完了3個(gè)GPU上的要更新的梯度， 計(jì)算一下平均梯度，然后更新參數(shù)。

然后繼續(xù)循環(huán)這個(gè)過(guò)程。

通過(guò)這個(gè)過(guò)程，處理的速度取決于最慢的那個(gè)GPU的速度。如果3個(gè)GPU的處理速度差不多的話， 處理速度就相當(dāng)于單機(jī)單GPU的速度的3倍減去數(shù)據(jù)在CPU和GPU之間傳輸?shù)拈_(kāi)銷(xiāo)，實(shí)際的效率提升看CPU和GPU之間數(shù)據(jù)的速度和處理數(shù)據(jù)的大小。

二 分布式TensorFlow

Tensorflow分布式訓(xùn)練的支持主要是通過(guò)tf.distribute.Strategy來(lái)實(shí)現(xiàn)

1 MirroredStrategy? 單機(jī)多卡訓(xùn)練

in-graph replication with synchronous

MirroredStrategy是一種支持多張GPU在同一個(gè)機(jī)器上的同步訓(xùn)練方法。在訓(xùn)練開(kāi)始時(shí)，Mirrored會(huì)在每張卡上復(fù)制一份模型，

每個(gè)顯卡會(huì)收到tf.data.Dataset傳來(lái)的數(shù)據(jù)，獨(dú)立計(jì)算梯度，然后采用all-reduce的方法進(jìn)行同步更新。多個(gè)顯卡在通信時(shí)默認(rèn)使用Nvidia NCCL進(jìn)行。

我們可以深入MirroredStrategy的實(shí)現(xiàn)了解一下?；旧纤械膁istributed strategy都是通過(guò)某些collective ops和cross device ops進(jìn)行數(shù)據(jù)通訊。MirroredStrategy也是如此，它是這樣選擇cross device ops的：

if len(workers) > 1:if not isinstance(self._cross_device_ops, cross_device_ops_lib.MultiWorkerAllReduce):raise ValueError("In-graph multi-worker training with `MirroredStrategy` is not ""supported.")self._inferred_cross_device_ops = self._cross_device_ops else:# TODO(yuefengz): make `choose_the_best` work with device strings# containing job names.self._inferred_cross_device_ops = cross_device_ops_lib.NcclAllReduce()

這也就印證了MirroredStrategy在單機(jī)多卡的情況下默認(rèn)使用NCCL來(lái)進(jìn)行通信的說(shuō)明。具體的實(shí)現(xiàn)大家可以去查看AllReduceCrossDeviceOps的實(shí)現(xiàn)。

同時(shí)，上面的程序也說(shuō)明MirroredStrategy可以運(yùn)用到多機(jī)多卡的情況中去，然而多機(jī)多卡的情況下用戶需要自己傳入cross_device_ops_lib.MultiWorkerAllReduce進(jìn)行通訊，這里MultiWorkerAllReduce支持若干種通訊方式，比如nccl,?nccl/xring,?nccl/rechd,?nccl/pscpu,?xring,?pscpu,?pscpu/pscpu等等。由于目前最佳的通訊方式需要NCCL2.0加上x(chóng)ring，然而Tensorflow目前使用NCCL 1.1，并且nccl/xring在現(xiàn)有的代碼中有bug無(wú)法工作，所以這一模式常常被大家詬病。

MirroredStrategy instance which will use all the GPUs that are visible to TensorFlow, and use NCCL as the cross device communication.

訓(xùn)練腳本就會(huì)自動(dòng)進(jìn)行分布式訓(xùn)練。如果你只想用主機(jī)上的部分GPU訓(xùn)練

mirrored_strategy = tf.distribute.MirroredStrategy(devices=["/gpu:0", "/gpu:1"])

用戶通過(guò)該API控制使用何種分布式架構(gòu)，例如如果用戶需要在單機(jī)多卡環(huán)境中使用All-Reduce架構(gòu)，只需定義對(duì)應(yīng)架構(gòu)下的Strategy，指定Estimator的config參數(shù)即可：

mirrored_strategy = tf.distribute.MirroredStrategy() config = tf.estimator.RunConfig(train_distribute=mirrored_strategy, eval_distribute=mirrored_strategy) regressor = tf.estimator.LinearRegressor(feature_columns=[tf.feature_column.numeric_column('feats')],optimizer='SGD',config=config)

tf.keras 例子

import tensorflow as tf import tensorflow_datasets as tfdsnum_epochs = 5 batch_size_per_replica = 64 learning_rate = 0.001strategy = tf.distribute.MirroredStrategy() print('Number of devices: %d' % strategy.num_replicas_in_sync) # 輸出設(shè)備數(shù)量 batch_size = batch_size_per_replica * strategy.num_replicas_in_sync# 載入數(shù)據(jù)集并預(yù)處理 def resize(image, label):image = tf.image.resize(image, [224, 224]) / 255.0return image, label# 當(dāng)as_supervised為T(mén)rue時(shí)，返回image和label兩個(gè)鍵值 dataset = tfds.load("cats_vs_dogs", split=tfds.Split.TRAIN, as_supervised=True) dataset = dataset.map(resize).shuffle(1024).batch(batch_size)with strategy.scope():model = tf.keras.applications.MobileNetV2()model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate),loss=tf.keras.losses.sparse_categorical_crossentropy,metrics=[tf.keras.metrics.sparse_categorical_accuracy])model.fit(dataset, epochs=num_epochs)

MirroredStrategy 的步驟如下：

訓(xùn)練開(kāi)始前，該策略在所有 N 個(gè)計(jì)算設(shè)備上均各復(fù)制一份完整的模型；

每次訓(xùn)練傳入一個(gè)批次的數(shù)據(jù)時(shí)，將數(shù)據(jù)分成 N 份，分別傳入 N 個(gè)計(jì)算設(shè)備（即數(shù)據(jù)并行）；

N 個(gè)計(jì)算設(shè)備使用本地變量（鏡像變量）分別計(jì)算自己所獲得的部分?jǐn)?shù)據(jù)的梯度；

使用分布式計(jì)算的 All-reduce 操作，在計(jì)算設(shè)備間高效交換梯度數(shù)據(jù)并進(jìn)行求和，使得最終每個(gè)設(shè)備都有了所有設(shè)備的梯度之和；

使用梯度求和的結(jié)果更新本地變量（鏡像變量）；

當(dāng)所有設(shè)備均更新本地變量后，進(jìn)行下一輪訓(xùn)練（即該并行策略是同步的）。

默認(rèn)情況下，TensorFlow 中的 MirroredStrategy 策略使用 NVIDIA NCCL 進(jìn)行 All-reduce 操作。

2? MultiWorkerMirroredStrategy 多機(jī)訓(xùn)練

對(duì)于分布式多機(jī)環(huán)境，最早是Uber專(zhuān)門(mén)提出了一種基于Ring-Allreduce的分布式TensorFlow架構(gòu)Horovod，并已開(kāi)源。

tf.distribute.experimental.MultiWorkerMirroredStrategy與MirroredStrategy非常類(lèi)似，都在每一個(gè)device上存儲(chǔ)一份模型的備份，進(jìn)行同步的分布式訓(xùn)練。

該策略采用CollectiveOps作為多個(gè)worker之間通訊的操作。所謂的collective op是Tensorflow自己實(shí)現(xiàn)的根據(jù)當(dāng)前硬件環(huán)境，網(wǎng)絡(luò)結(jié)構(gòu)，和Tensor大小自動(dòng)采用最佳算法進(jìn)行all-reduce的計(jì)算操作。一個(gè)collective op的實(shí)現(xiàn)邏輯十分簡(jiǎn)單

if (CanProceedWithCompute(c, col_exec, done)) {col_exec->ExecuteAsync(c, col_params_, GetCollectiveKey(c), actual_done); }

c是當(dāng)前op的計(jì)算狀態(tài)，col_exec是Tensorflow根據(jù)系統(tǒng)情況選擇的collective executor，所有的all reduce，boardcast和receive操作都有collective executor去執(zhí)行。

該策略目前也實(shí)現(xiàn)了很多優(yōu)化，比如將很多個(gè)小tensor的all reduce操作變成幾個(gè)大tensor的all reduce操作，以及在開(kāi)發(fā)當(dāng)中的采用最新NCCL 2.0進(jìn)行通訊的操作，具體可以參見(jiàn)Issue 24505?？梢钥闯鯰ensorflow分布式訓(xùn)練在被吐槽很多次后，感受到了來(lái)自Pytorch，Horovod的壓力，在努力的提升自己。

最后，關(guān)于MultiWorkerMirroredStrategy的配置，有兩點(diǎn)需要注意。

一點(diǎn)是collective ops的策略選擇，目前支持CollectiveCommunication.RING，采用與Horovod類(lèi)似的ring-based通訊策略。另一個(gè)是CollectiveCommunication.NCCL，采用Nvidia NCCL進(jìn)行通訊，在啟動(dòng)策略時(shí)可以傳入?yún)?shù)指定：

multiworker_strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy(tf.distribute.experimental.CollectiveCommunication.NCCL)

CollectiveCommunication.AUTO defers the choice to the runtime.

另一個(gè)需要注意的是關(guān)于TF_CONFIG的設(shè)置，該策略并不需要指定Parameter server，只需要一系列worker即可，其配置如下：

TF_CONFIG = {'cluster': {'worker': ['worker1:port1', 'worker2:port2', 'worker3:port3', ...]},'task': {'type': 'worker', 'index': 0} })

目前該API尚處于實(shí)驗(yàn)階段。如果在代碼中通過(guò)MultiWorkerMirroredStrategy指定使用All-Reduce架構(gòu)，則分布式提交時(shí)，TF_CONFIG環(huán)境變量中的cluster就不需要ps類(lèi)型的節(jié)點(diǎn)了，例如：

TF_CONFIG='{"cluster": {"worker": ["host1:2222", "host2:2222", "host3:2222"]},"task": {"type": "work", "index": 0} }'

strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy() config = tf.estimator.RunConfig(train_distribute=strategy, eval_distribute=strategy) regressor = tf.estimator.LinearRegressor(feature_columns=[tf.feature_column.numeric_column('feats')],optimizer='SGD',config=config)

多機(jī)訓(xùn)練的方法和單機(jī)多卡類(lèi)似，將 MirroredStrategy 更換為適合多機(jī)訓(xùn)練的 MultiWorkerMirroredStrategy 即可。不過(guò)，由于涉及到多臺(tái)計(jì)算機(jī)之間的通訊，還需要進(jìn)行一些額外的設(shè)置。具體而言，需要設(shè)置環(huán)境變量 TF_CONFIG ，示例如下:

os.environ['TF_CONFIG'] = json.dumps({'cluster': {'worker': ["localhost:20000", "localhost:20001"]},'task': {'type': 'worker', 'index': 0} })

TF_CONFIG 由 cluster 和 task 兩部分組成：

• cluster 說(shuō)明了整個(gè)多機(jī)集群的結(jié)構(gòu)和每臺(tái)機(jī)器的網(wǎng)絡(luò)地址（IP + 端口號(hào)）。對(duì)于每一臺(tái)機(jī)器，cluster 的值都是相同的；

• task 說(shuō)明了當(dāng)前機(jī)器的角色。例如， {'type': 'worker', 'index': 0} 說(shuō)明當(dāng)前機(jī)器是 cluster 中的第 0 個(gè) worker（即 localhost:20000 ）。每一臺(tái)機(jī)器的 task 值都需要針對(duì)當(dāng)前主機(jī)進(jìn)行分別的設(shè)置。

以上內(nèi)容設(shè)置完成后，在所有的機(jī)器上逐個(gè)運(yùn)行訓(xùn)練代碼即可。先運(yùn)行的代碼在尚未與其他主機(jī)連接時(shí)會(huì)進(jìn)入監(jiān)聽(tīng)狀態(tài)，待整個(gè)集群的連接建立完畢后，所有的機(jī)器即會(huì)同時(shí)開(kāi)始訓(xùn)練。

請(qǐng)?jiān)诟髋_(tái)機(jī)器上均注意防火墻的設(shè)置，尤其是需要開(kāi)放與其他主機(jī)通信的端口。如上例的 0 號(hào) worker 需要開(kāi)放 20000 端口，1 號(hào) worker 需要開(kāi)放 20001 端口。

以下示例的訓(xùn)練任務(wù)與前節(jié)相同，只不過(guò)遷移到了多機(jī)訓(xùn)練環(huán)境。假設(shè)我們有兩臺(tái)機(jī)器，即首先在兩臺(tái)機(jī)器上均部署下面的程序，唯一的區(qū)別是 task 部分，第一臺(tái)機(jī)器設(shè)置為 {'type': 'worker', 'index': 0} ，第二臺(tái)機(jī)器設(shè)置為 {'type': 'worker', 'index': 1} 。接下來(lái)，在兩臺(tái)機(jī)器上依次運(yùn)行程序，待通訊成功后，即會(huì)自動(dòng)開(kāi)始訓(xùn)練流程。

tf.keras例子

import tensorflow as tf import tensorflow_datasets as tfds import os import jsonnum_epochs = 5 batch_size_per_replica = 64 learning_rate = 0.001num_workers = 2 os.environ['TF_CONFIG'] = json.dumps({'cluster': {'worker': ["localhost:20000", "localhost:20001"]},'task': {'type': 'worker', 'index': 0} }) strategy = tf.distribute.experimental.MultiWorkerMirroredStrategy() batch_size = batch_size_per_replica * num_workersdef resize(image, label):image = tf.image.resize(image, [224, 224]) / 255.0return image, labeldataset = tfds.load("cats_vs_dogs", split=tfds.Split.TRAIN, as_supervised=True) dataset = dataset.map(resize).shuffle(1024).batch(batch_size)with strategy.scope():model = tf.keras.applications.MobileNetV2()model.compile(optimizer=tf.keras.optimizers.Adam(learning_rate=learning_rate),loss=tf.keras.losses.sparse_categorical_crossentropy,metrics=[tf.keras.metrics.sparse_categorical_accuracy])model.fit(dataset, epochs=num_epochs)

3. CentralStorageStrategy

tf.distribute.experimental.CentralStorageStrategy也執(zhí)行同步訓(xùn)練，但是變量不會(huì)被鏡像，而是放在CPU上。各操作(operation)在本地GPU之間復(fù)制進(jìn)行。如果只有一個(gè)GPU，變量和操作都會(huì)放在GPU上。

創(chuàng)建一個(gè) CentralStorageStrategy 實(shí)例：

central_storage_strategy = tf.distribute.experimental.CentralStorageStrategy() INFO:tensorflow:ParameterServerStrategy (CentralStorageStrategy if you are using a single machine) with compute_devices = ['/job:localhost/replica:0/task:0/device:GPU:0'], variable_device = '/job:localhost/replica:0/task:0/device:GPU:0'

這會(huì)創(chuàng)建一個(gè) CentralStorageStrategy 實(shí)例使用所有可見(jiàn)的CPU和GPU。在更新應(yīng)用到變量之前，不同副本上變量的更新將會(huì)匯總。

注意: 該策略是 實(shí)驗(yàn)性的 ，因?yàn)槲覀冋趯?duì)它進(jìn)行改進(jìn)，使他能在更多場(chǎng)景下工作. 敬請(qǐng)期待此API的變化。

4 ParameterServerStrategy

他是Tensorflow最初的分布式訓(xùn)練方法，它由若干個(gè)parameter servers和若干個(gè)worker servers構(gòu)成，parameter servers用于存儲(chǔ)參數(shù)，workers用于計(jì)算。

ps_strategy = tf.distribute.experimental.ParameterServerStrategy()

ParameterServerStrategy 在訓(xùn)練過(guò)程中worker servers會(huì)和不同的parameter servers溝通獲得參數(shù)，然后計(jì)算，向parameter servers傳遞參數(shù)的梯度。配置一個(gè)這樣的訓(xùn)練環(huán)境非常簡(jiǎn)單，只需要在程序運(yùn)行時(shí)設(shè)置好環(huán)境變量TF_CONFIG，需要注意的是需要給分布式集群里每一個(gè)機(jī)子不同的task。

os.environ["TF_CONFIG"] = json.dumps({"cluster": {"worker": ["host1:port", "host2:port", "host3:port"],"ps": ["host4:port", "host5:port"]},"task": {"type": "worker", "index": 1} })

同時(shí)，ParameterServerStrategy還有比較神奇的功能，它可以通過(guò)傳入num_gpus_per_worker在一個(gè)worker上進(jìn)行多GPU的同步計(jì)算，然后不同worker之間進(jìn)行異步計(jì)算。但是由于單一worker上多GPU并沒(méi)有利用NCCL進(jìn)行通訊，而是直接將結(jié)果發(fā)送到CPU，所以效率非常低下。

strategy = tf.distribute.experimental.ParameterServerStrategy() run_config = tf.estimator.RunConfig(experimental_distribute.train_distribute=strategy) estimator = tf.estimator.Estimator(config=run_config) tf.estimator.train_and_evaluate(estimator,...)

Examples and Tutorials

Here is a list of tutorials and examples that illustrate the above integration end to end with Keras:

Tutorial?to train MNIST using?MirroredStrategy.

Tutorial?to train MNIST using?MultiWorkerMirroredStrategy.

DenseNet?example using?MirroredStrategy.

BERT?example trained using?MirroredStrategy?and?TPUStrategy. This example is particularly helpful for understanding how to load from a checkpoint and generate periodic checkpoints during distributed training etc.

NCF?example trained using?MirroredStrategy?and?TPUStrategy?that can be enabled using the?keras_use_ctl?flag.

Transformer?trained using?MirroredStrategy.

NMT?example trained using?MirroredStrategy.

Official?ResNet50?training with ImageNet data using?MirroredStrategy.

ResNet50?trained with Imagenet data on Cloud TPus with?TPUStrategy.

We've integrated tf.distribute.Strategy into tf.keras which is TensorFlow's implementation of the Keras API specification.? tf.keras is a high-level API to build and train models. By integrating into tf.keras backend, we've made it seamless for Keras users to distribute their training written in the Keras training framework.

The only things that need to change in a user's program are:

(1) Create an instance of the appropriate tf.distribute.Strategy

(2) Move the creation and compiling of Keras model inside strategy.scope.

Here is a snippet of code to do this for a very simple Keras model with one dense layer:

mirrored_strategy = tf.distribute.MirroredStrategy() with mirrored_strategy.scope():model = tf.keras.Sequential([tf.keras.layers.Dense(1, input_shape=(1,))])model.compile(loss='mse', optimizer='sgd')

三? 分布式訓(xùn)練的容錯(cuò) Fault tolerance

在同步訓(xùn)練中, 如果有一個(gè)worker 失敗了, 整個(gè)訓(xùn)練集群就會(huì)失敗，沒(méi)有故障恢復(fù)機(jī)制.

Using Keras with?tf.distribute.Strategy?comes with the advantage of fault tolerance in cases where workers die or are otherwise unstable. We do this by preserving training state in the distributed file system of your choice, such that upon restart of the instance that previously failed or preempted, the training state is recovered.

Since all the workers are kept in sync in terms of training epochs and steps, other workers would need to wait for the failed or preempted worker to restart to continue.

ModelCheckpoint callback

To take advantage of fault tolerance in multi-worker training, provide an instance of?tf.keras.callbacks.ModelCheckpoint?at the?tf.keras.Model.fit()?call. The callback will store the checkpoint and training state in the directory corresponding to the?filepath?argument to?ModelCheckpoint.

# Replace the `filepath` argument with a path in the file system # accessible by all workers. callbacks = [tf.keras.callbacks.ModelCheckpoint(filepath='/tmp/keras-ckpt')] with strategy.scope():multi_worker_model = build_and_compile_cnn_model() multi_worker_model.fit(x=train_datasets, epochs=3, callbacks=callbacks) Epoch 1/3469/Unknown - 8s 18ms/step - loss: 2.2049 - accuracy: 0.2318WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.6/site-packages/tensorflow_core/python/ops/resource_variable_ops.py:1781: calling BaseResourceVariable.__init__ (from tensorflow.python.ops.resource_variable_ops) with constraint is deprecated and will be removed in a future version. Instructions for updating: If using Keras pass *_constraint arguments to layers.WARNING:tensorflow:From /tmpfs/src/tf_docs_env/lib/python3.6/site-packages/tensorflow_core/python/ops/resource_variable_ops.py:1781: calling BaseResourceVariable.__init__ (from tensorflow.python.ops.resource_variable_ops) with constraint is deprecated and will be removed in a future version. Instructions for updating: If using Keras pass *_constraint arguments to layers.INFO:tensorflow:Assets written to: /tmp/keras-ckpt/assetsINFO:tensorflow:Assets written to: /tmp/keras-ckpt/assets469/469 [==============================] - 9s 19ms/step - loss: 2.2049 - accuracy: 0.2318 Epoch 2/3 451/469 [===========================>..] - ETA: 0s - loss: 1.9195 - accuracy: 0.5715INFO:tensorflow:Assets written to: /tmp/keras-ckpt/assetsINFO:tensorflow:Assets written to: /tmp/keras-ckpt/assets469/469 [==============================] - 2s 4ms/step - loss: 1.9113 - accuracy: 0.5767 Epoch 3/3 450/469 [===========================>..] - ETA: 0s - loss: 1.4175 - accuracy: 0.7550INFO:tensorflow:Assets written to: /tmp/keras-ckpt/assetsINFO:tensorflow:Assets written to: /tmp/keras-ckpt/assets469/469 [==============================] - 2s 4ms/step - loss: 1.4078 - accuracy: 0.7561<tensorflow.python.keras.callbacks.History at 0x7fc38fdfee80> ?

If a worker gets preempted, the whole cluster pauses until the preempted worker is restarted. Once the worker rejoins the cluster, other workers will also restart. Now, every worker reads the checkpoint file that was previously saved and picks up its former state, thereby allowing the cluster to get back in sync. Then the training continues.

If you inspect the directory containing the?filepath?you specified in?ModelCheckpoint, you may notice some temporarily generated checkpoint files. Those files are needed for recovering the previously lost instances, and they will be removed by the library at the end of?tf.keras.Model.fit()?upon successful exiting of your multi-worker training.

三 總結(jié)

本文梳理了分布式TensorFlow編程模型的發(fā)展，主要從用戶使用分布式TensorFlow角度出發(fā)，闡述了不同的分布式TensorFlow架構(gòu)?？梢钥吹?#xff0c;隨著TensorFlow的迭代演進(jìn)，其易用性越來(lái)越友好。目前TensorFlow已經(jīng)發(fā)布了2.0.0 正式版本，標(biāo)志著TensorFlow正式進(jìn)入2.0時(shí)代了，通過(guò)不同的Strategy，可以輕松控制使用不同的分布式TensorFlow架構(gòu)，可見(jiàn)TensorFlow的API設(shè)計(jì)更加靈活友好，擁有極強(qiáng)的可擴(kuò)展性，相信將來(lái)會(huì)出現(xiàn)更多的Strategy來(lái)應(yīng)對(duì)復(fù)雜的分布式場(chǎng)景。

在2.0版本中，其主打賣(mài)點(diǎn)是Eager Execution與Keras高階API，整體易用性將進(jìn)一步提升，通過(guò)Eager Execution功能，我們可以像使用原生Python一樣操作Tensor，而不需要像以前一樣需要通過(guò)Session.run的方式求解Tensor，另外，通過(guò)TensorFlow Keras高階API，可以更加靈活方便構(gòu)建模型，同時(shí)可以將模型導(dǎo)出為Keras標(biāo)準(zhǔn)格式HDF5，以靈活兼容在線服務(wù)等。

補(bǔ)充： Tensorflow 1.0---in-graph 和? between-graph

in-graph模式

In-graph模式，單機(jī)多GPU模型有點(diǎn)類(lèi)似，? 把計(jì)算已經(jīng)從單機(jī)多GPU，已經(jīng)擴(kuò)展到了多機(jī)多GPU了， 不過(guò)數(shù)據(jù)分發(fā)還是在一個(gè)節(jié)點(diǎn)，其他結(jié)算節(jié)點(diǎn)只需join操作。 這樣的好處是配置簡(jiǎn)單， 其他多機(jī)多GPU的計(jì)算節(jié)點(diǎn)，只要起個(gè)join操作， 暴露一個(gè)網(wǎng)絡(luò)接口，等在那里接受任務(wù)就好了。 這些計(jì)算節(jié)點(diǎn)暴露出來(lái)的網(wǎng)絡(luò)接口，使用起來(lái)就跟本機(jī)的一個(gè)GPU的使用一樣， 只要在操作的時(shí)候指定tf.device("/job:worker/task:N")，就可以向指定GPU一樣，把操作指定到一個(gè)計(jì)算節(jié)點(diǎn)上計(jì)算，使用起來(lái)和多GPU的類(lèi)似。 但是這樣的壞處是訓(xùn)練數(shù)據(jù)的分發(fā)依然在一個(gè)節(jié)點(diǎn)上， 要把訓(xùn)練數(shù)據(jù)分發(fā)到不同的機(jī)器上， 嚴(yán)重影響并發(fā)訓(xùn)練速度。在大數(shù)據(jù)訓(xùn)練的情況下， 不推薦使用這種模式。

對(duì)于圖內(nèi)復(fù)制，只構(gòu)建一個(gè)Client，這個(gè)Client構(gòu)建一個(gè)Graph，Graph中包含一套模型參數(shù)，放置在ps上，同時(shí)Graph中包含模型計(jì)算部分的多個(gè)副本，每個(gè)副本都放置在一個(gè)worker上，這樣多個(gè)worker可以同時(shí)訓(xùn)練復(fù)制的模型。

再開(kāi)一個(gè)Python解釋器，作為Client，執(zhí)行如下語(yǔ)句構(gòu)建計(jì)算圖，并：

import tensorflow as tfwith tf.device("/job:ps/task:0"):w = tf.get_variable([[1., 2., 3.], [1., 3., 5.]])input_data = ... inputs = tf.split(input_data, num_workers) outputs = []for i in range(num_workers):with tf.device("/job:ps/task:%s" % str(i)):outputs.append(tf.matmul(inputs[i], w))output = tf.concat(outputs, axis=0) with tf.Session() as sess:sess.run(tf.global_variables_initializer())print sess.run(output)

從以上代碼可以看到，當(dāng)采用圖內(nèi)復(fù)制時(shí)，需要在Client上創(chuàng)建一個(gè)包含所有worker副本的流程圖，隨著worker數(shù)量的增長(zhǎng)，計(jì)算圖將會(huì)變得非常大，不利于計(jì)算圖的維護(hù)。此外，數(shù)據(jù)分發(fā)在Client單點(diǎn)，要把訓(xùn)練數(shù)據(jù)分發(fā)到不同的機(jī)器上，會(huì)嚴(yán)重影響并發(fā)訓(xùn)練速度。所以在大規(guī)模分布式多機(jī)訓(xùn)練情況下，一般不會(huì)采用圖內(nèi)復(fù)制的模式，該模式常用于單機(jī)多卡情況下，簡(jiǎn)單直接。

between-graph模式

between-graph模式下，訓(xùn)練的參數(shù)保存在參數(shù)服務(wù)器，數(shù)據(jù)不用分發(fā)，數(shù)據(jù)分片的保存在各個(gè)計(jì)算節(jié)點(diǎn)，各個(gè)計(jì)算節(jié)點(diǎn)自己算自己的，算完后把要更新的參數(shù)告訴參數(shù)服務(wù)器，參數(shù)服務(wù)器更新參數(shù)。這種模式的優(yōu)點(diǎn)是不用進(jìn)行訓(xùn)練數(shù)據(jù)的分發(fā)，尤其數(shù)據(jù)量在TB級(jí)的時(shí)候，節(jié)省了大量的時(shí)間，所以大數(shù)據(jù)深度學(xué)習(xí)推薦使用between-graph模式。

為可以解決圖內(nèi)復(fù)制在擴(kuò)展上的局限性，我們可以采用圖間復(fù)制模式。對(duì)于圖間復(fù)制，每個(gè)worker節(jié)點(diǎn)上都創(chuàng)建一個(gè)Client，各個(gè)Client構(gòu)建相同的Graph，但是參數(shù)還是放置在ps上，每個(gè)worker節(jié)點(diǎn)單獨(dú)運(yùn)算，一個(gè)worker節(jié)點(diǎn)掛掉了，系統(tǒng)還可以繼續(xù)跑。

所以我們?cè)诘谝粋€(gè)worker和第二個(gè)worker的Python解釋器里繼續(xù)執(zhí)行如下語(yǔ)句實(shí)現(xiàn)Client完成整個(gè)分布式TensorFlow的運(yùn)行：

with tf.device("/job:ps/task:0"):w = tf.get_variable(name='w', shape=[784, 10])b = tf.get_variable(name='b', shape=[10])x = tf.placeholder(tf.float32, shape=[None, 784]) y = tf.placeholder(tf.int32, shape=[None]) logits = tf.matmul(x, w) + b loss = ... train_op = ...with tf.Session() as sess:for _ in range(10000):sess.run(train_op, feed_dict=...)

在上述描述的過(guò)程中，我們是全程手動(dòng)做分布式驅(qū)動(dòng)的，先建立Cluster，然后構(gòu)建計(jì)算圖提交執(zhí)行，Server上的Master Service和Worker Service根本沒(méi)有用到。實(shí)際應(yīng)用時(shí)當(dāng)然不會(huì)這么愚蠢，一般是將以上代碼片段放到一個(gè)文件中，通過(guò)參數(shù)控制執(zhí)行不同的代碼片段，例如：

import tensorflow as tfps_hosts = FLAGS.ps_hosts.split(",") worker_hosts = FLAGS.worker_hosts.split(",") cluster = tf.train.ClusterSpec({"ps": ps_hosts, "worker": worker_hosts}) server = tf.train.Server(cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index)if FLAGS.job_name == 'ps':server.join() elif FLAGS.job_name == "worker":with tf.device(tf.train.replica_device_setter(worker_device="/job:worker/task:%d" % FLAGS.task_index,cluster=cluster)):# Build model...loss = ...train_op = ...with tf.train.MonitoredTrainingSession(master="/job:worker/task:0",is_chief=(FLAGS.task_index == 0),checkpoint_dir="/tmp/train_logs") as mon_sess:while not mon_sess.should_stop():mon_sess.run(train_op)

每個(gè)節(jié)點(diǎn)上都執(zhí)行如上代碼，只是不同節(jié)點(diǎn)輸入的參數(shù)不一樣，對(duì)于ps節(jié)點(diǎn)，啟動(dòng)Server后就堵塞等待參數(shù)服務(wù)，對(duì)于worker節(jié)點(diǎn)，啟動(dòng)Server后(后臺(tái)服務(wù))，開(kāi)始扮演Client，構(gòu)建計(jì)算圖，最后通過(guò)Session提交計(jì)算。注意在調(diào)用Session.run之前，僅僅是Client的構(gòu)圖，并未開(kāi)始計(jì)算，各節(jié)點(diǎn)上的Server還未發(fā)揮作用，只有在調(diào)用Session.run后，worker和ps節(jié)點(diǎn)才會(huì)被派發(fā)Task。在調(diào)用Session.run時(shí)，需要給Session傳遞target參數(shù)，指定使用哪個(gè)worker節(jié)點(diǎn)上的Master Service，Client將構(gòu)建的計(jì)算圖發(fā)給target指定的Master Service，一個(gè)TensorFlow集群中只有一個(gè)Master Service在工作，它負(fù)責(zé)子圖劃分、Task的分發(fā)以及模型保存與恢復(fù)等，在子圖劃分時(shí)，它會(huì)自動(dòng)將模型參數(shù)分發(fā)到ps節(jié)點(diǎn)，將梯度計(jì)算分發(fā)到worker節(jié)點(diǎn)。另外，在Client構(gòu)圖時(shí)通過(guò)tf.train.replica_device_setter告訴worker節(jié)點(diǎn)默認(rèn)在本機(jī)分配Op，這樣每個(gè)Worker Service收到計(jì)算任務(wù)后構(gòu)建出一個(gè)單獨(dú)的計(jì)算子圖副本，這樣每個(gè)worker節(jié)點(diǎn)就可以單獨(dú)運(yùn)行，掛了不影響其他worker節(jié)點(diǎn)繼續(xù)運(yùn)行。

雖然圖間復(fù)制具有較好的擴(kuò)展性，但是從以上代碼可以看到，寫(xiě)一個(gè)分布式TensorFlow應(yīng)用，需要用戶自行控制不同組件的運(yùn)行，這就需要用戶對(duì)TensorFlow的分布式架構(gòu)有較深的理解。另外，分布式TensorFlow應(yīng)用與單機(jī)版TensorFlow應(yīng)用的代碼是兩套，一般使用過(guò)程中，用戶都是先在單機(jī)上調(diào)試好基本邏輯，然后再部署到集群，在部署分布式TensorFlow應(yīng)用前，就需要將前面的單機(jī)版代碼改寫(xiě)成分布式多機(jī)版，用戶體驗(yàn)非常差。所以說(shuō)，使用Low-level 分布式編程模型，不能做到一套代碼既可以在單機(jī)上運(yùn)行也可以在分布式多機(jī)上運(yùn)行，其用戶門(mén)檻較高，一度被相關(guān)工程及研究人員詬病。為此，TensorFlow推出了High-level分布式編程模型，極大地改善用戶易用性。

同步更新和異步更新

in-graph和between-graph模式都支持同步更新和異步更新。

在同步更新的時(shí)候，每次梯度更新，要等所有分發(fā)的數(shù)據(jù)計(jì)算完成，返回結(jié)果，把梯度累加算了均值之后，再更新參數(shù)。這樣的好處是loss的下降比較穩(wěn)定，但這個(gè)的壞處也比較明顯，處理的速度取決于最慢的那個(gè)分片的計(jì)算時(shí)間。

在異步更新時(shí)，所有的計(jì)算節(jié)點(diǎn)，自己算自己的，更新參數(shù)也是自己更新自己的計(jì)算結(jié)果，這樣的優(yōu)點(diǎn)是計(jì)算速度快，計(jì)算資源能得到充分利用，但是缺點(diǎn)是loss的下降不穩(wěn)定，抖動(dòng)大。

在數(shù)據(jù)量小的情況下，各個(gè)節(jié)點(diǎn)的計(jì)算能力比較均衡的情況下，推薦使用同步模式；數(shù)據(jù)量很大，各個(gè)機(jī)器的計(jì)算性能參差不齊的情況下，推薦使用異步的方式。

TensorFlow 1.X 版本的分布式

最原始的分布式TensorFlow

Parameter Server的配置數(shù)量也非常復(fù)雜，不同的網(wǎng)絡(luò)環(huán)境，模型大小都會(huì)對(duì)效率有影響，所以現(xiàn)在官方好像也不怎么推薦這種做法了。最原始的分布式TensorFlow編程是基于Low-level API來(lái)實(shí)現(xiàn)，下面我們通過(guò)舉例來(lái)理解最原始的分布式TensorFlow編程步驟。我們?cè)谝慌_(tái)機(jī)器上啟動(dòng)三個(gè)Server(2個(gè)worker，1個(gè)ps)來(lái)模擬分布式多機(jī)環(huán)境，開(kāi)啟三個(gè)Python解釋器(分別對(duì)應(yīng)2個(gè)worker和1個(gè)ps)，執(zhí)行如下python語(yǔ)句，定義一個(gè)Cluster：

import tensorflow as tfcluster = tf.train.ClusterSpec({"worker": ["localhost:2222","localhost:2223"],"ps": ["localhost:2224"]})在第一個(gè)worker解釋器內(nèi)執(zhí)行如下語(yǔ)句啟動(dòng)Server： server = tf.train.Server(cluster, job_name="worker", task_index=0)

在第二個(gè)worker解釋器內(nèi)執(zhí)行如下語(yǔ)句啟動(dòng)Server：

server = tf.train.Server(cluster, job_name="worker", task_index=1)在ps解釋器內(nèi)執(zhí)行如下語(yǔ)句啟動(dòng)Server: server = tf.train.Server(cluster, job_name="ps", task_index=0)

至此，我們已經(jīng)啟動(dòng)了一個(gè)TensorFlow Cluster，它由兩個(gè)worker節(jié)點(diǎn)和一個(gè)ps節(jié)點(diǎn)組成，每個(gè)節(jié)點(diǎn)上都有Master Service和Worker Service，其中worker節(jié)點(diǎn)上的Worker Service將負(fù)責(zé)梯度運(yùn)算，ps節(jié)點(diǎn)上的Worker Service將負(fù)責(zé)參數(shù)更新，三個(gè)Master Service將僅有一個(gè)會(huì)在需要時(shí)被用到，負(fù)責(zé)子圖劃分與Task派發(fā)。

上圖所示，假設(shè)存在兩個(gè)任務(wù)：

• /job:ps/task:0: 負(fù)責(zé)模型參數(shù)的存儲(chǔ)和更新
• /job:worker/task:0: 負(fù)責(zé)模型的訓(xùn)練或推理

有了Cluster，我們就可以編寫(xiě)Client，構(gòu)建計(jì)算圖，并提交到這個(gè)Cluster上執(zhí)行。使用分布式TensorFlow時(shí)，最常采用的分布式訓(xùn)練策略是數(shù)據(jù)并行，數(shù)據(jù)并行就是在很多設(shè)備上放置相同的模型，在TensorFlow中稱(chēng)之為Replicated training，主要表現(xiàn)為兩種模式：圖內(nèi)復(fù)制(in-graph replication)和圖間復(fù)制(between-graph replication)。不同的運(yùn)行模式，Client的表現(xiàn)形式不一樣。

• Client
  可以把它看成是TensorFlow前端，它支持多語(yǔ)言的編程環(huán)境(Python/C++/Go/Java等)，方便用戶構(gòu)造各種復(fù)雜的計(jì)算圖。Client通過(guò)Session連接TensorFlow后端，并啟動(dòng)計(jì)算圖的執(zhí)行。Client基于TensorFlow的編程接口，構(gòu)造計(jì)算圖。此時(shí)，TensorFlow并未執(zhí)行任何計(jì)算。直至建立Session會(huì)話，并以Session為橋梁，建立Client與后端運(yùn)行時(shí)的通道，將Protobuf格式的GraphDef發(fā)送至Distributed Master。也就是說(shuō)，當(dāng)Client對(duì)OP結(jié)果進(jìn)行求值時(shí)，將觸發(fā)Distributed Master的計(jì)算圖的執(zhí)行過(guò)程

• Master
  Master根據(jù)要計(jì)算的操作(Op)，從計(jì)算圖中反向遍歷，找到其所依賴的最小子圖，然后將該子圖再次分裂為多個(gè)子圖片段，以便在不同的進(jìn)程和設(shè)備上運(yùn)行這些子圖片段，最后將這些子圖片段派發(fā)給Worker執(zhí)行。

• Worker
  Worker按照計(jì)算子圖中節(jié)點(diǎn)之間的依賴關(guān)系，根據(jù)當(dāng)前的可用的硬件環(huán)境(GPU/CPU/TPU)，調(diào)用Op的Kernel實(shí)現(xiàn)完成運(yùn)算。對(duì)于每個(gè)任務(wù)，都將存在相應(yīng)的Worker Service，它主要負(fù)責(zé)如下3個(gè)方面的職責(zé)：1 處理來(lái)自Master的請(qǐng)求；2 調(diào)度OP的Kernel實(shí)現(xiàn)，執(zhí)行本地子圖；3 協(xié)同任務(wù)之間的數(shù)據(jù)通信。

在分布式TensorFlow中，參與分布式系統(tǒng)的所有節(jié)點(diǎn)或者設(shè)備統(tǒng)稱(chēng)為一個(gè)Cluster，一個(gè)Cluster中包含很多Server，每個(gè)Server去執(zhí)行一項(xiàng)Task，Server和Task是一一對(duì)應(yīng)的。

所以，Cluster可以看成是Server的集合，也可以看成是Task的集合，TensorFlow為各個(gè)Task又增加了一個(gè)抽象層，將一系列相似的Task集合稱(chēng)為一個(gè)Job。

一組Task集合(即Job)有若干個(gè)Server(host和port標(biāo)識(shí))，每個(gè)Server上會(huì)綁定兩個(gè)Service，就是前面提到的Master Service和Worker Service，Client通過(guò)Session連接集群中的任意一個(gè)Server的Master Service提交計(jì)算圖，Master Service負(fù)責(zé)劃分子圖并派發(fā)Task給Worker Service，Worker Service則負(fù)責(zé)運(yùn)算派發(fā)過(guò)來(lái)的Task完成子圖的運(yùn)算。

為什么要分成Cluster Job和Task

首先,我們介紹一下Task:Task就是主機(jī)上的一個(gè)進(jìn)程,在大多數(shù)情況下,一個(gè)機(jī)器上只運(yùn)行一個(gè)Task.

為什么Job是Task的集合呢? 在分布式深度學(xué)習(xí)框架中,我們一般把Job劃分為Parameter和Worker,Parameter Job是管理參數(shù)的存儲(chǔ)和更新工作.Worker Job是來(lái)運(yùn)行ops.如果參數(shù)的數(shù)量太大,一臺(tái)機(jī)器處理不了,這就要需要多個(gè)Tasks.

Cluster?是?Jobs?的集合:?Cluster(集群),就是我們用的集群系統(tǒng)了

參數(shù)服務(wù)器

當(dāng)計(jì)算模型越來(lái)越大，模型的參數(shù)越來(lái)越多，多到模型參數(shù)的更新，一臺(tái)機(jī)器的性能都不夠時(shí)，我們需要將參數(shù)分開(kāi)到不同的機(jī)器去存儲(chǔ)和更新。參數(shù)服務(wù)器可以是多臺(tái)機(jī)器組成的集群，類(lèi)似于分布式的存儲(chǔ)結(jié)構(gòu)。主要用來(lái)解決參數(shù)存儲(chǔ)和更新的性能問(wèn)題。

對(duì)于PS架構(gòu)，Parameter Server的Task集合為ps(即job類(lèi)型為ps)，而執(zhí)行梯度計(jì)算的Task集合為worker(即job類(lèi)型為worker)，Low-level 分布式編程模型

High-level 分布式編程模型

TensorFlow提供Estimator和Dataset高階API，簡(jiǎn)化模型構(gòu)建以及數(shù)據(jù)輸入，用戶通過(guò)Estimator和Dataset高階API編寫(xiě)TensorFlow應(yīng)用，不用了解TensorFlow內(nèi)部實(shí)現(xiàn)細(xì)節(jié)，只需關(guān)注模型本身即可。

Estimator代表一個(gè)完整的模型，它提供方法用于模型的訓(xùn)練、評(píng)估、預(yù)測(cè)及導(dǎo)出

Estimator具備如下優(yōu)勢(shì)：

• 基于Estimator編寫(xiě)的代碼，可運(yùn)行在單機(jī)和分布式環(huán)境中，不用區(qū)別對(duì)待
• 簡(jiǎn)化了模型開(kāi)發(fā)者之間共享部署，它提供了標(biāo)準(zhǔn)的模型導(dǎo)出功能，可以將訓(xùn)練好的模型直接用于TensorFlow-Serving等在線服務(wù)
• 提供全套的分布式訓(xùn)練生命周期管理，自動(dòng)初始化變量、處理異常、創(chuàng)建檢查點(diǎn)文件并從故障中恢復(fù)、以及保存TensorBoard 的摘要等
• 提供了一系列開(kāi)箱即用的常見(jiàn)Estimator，例如DNNClassifier，LinearClassifier等

使用Estimator編寫(xiě)應(yīng)用時(shí)，需將數(shù)據(jù)輸入從模型中分離出來(lái)。數(shù)據(jù)輸入可以通過(guò)?Dataset?API 構(gòu)建數(shù)據(jù) pipeline，類(lèi)似Spark RDD或DataFrame，可以輕松處理大規(guī)模數(shù)據(jù)、不同的數(shù)據(jù)格式以及復(fù)雜的轉(zhuǎn)換等。具體關(guān)于Estimator的使用可以參考TensorFlow官方文檔，講的特別詳細(xì)。

使用Estimator編寫(xiě)完應(yīng)用后，可以直接單機(jī)上運(yùn)行，如果需要將其部署到分布式環(huán)境運(yùn)行，則需要在每個(gè)節(jié)點(diǎn)執(zhí)行代碼前設(shè)置集群的TF_CONFIG環(huán)境變量(實(shí)際應(yīng)用時(shí)通常借助資源調(diào)度平臺(tái)自動(dòng)完成，如K8S，不需要修改TensorFlow應(yīng)用程序代碼)：

TF_CONFIG='{"cluster": {"chief": ["host0:2222"],"worker": ["host1:2222", "host2:2222", "host3:2222"],"ps": ["host4:2222", "host5:2222"]},"task": {"type": "chief", "index": 0} }'

TF_CONFIG環(huán)境變量是一個(gè)json字符串，指定集群規(guī)格cluster以及節(jié)點(diǎn)自身的角色task，cluster包括chief、worker、ps節(jié)點(diǎn)，chief節(jié)點(diǎn)其實(shí)是一個(gè)特殊的worker節(jié)點(diǎn)，而且只能有一個(gè)節(jié)點(diǎn)，表示分布式TensorFlow Master Service所在的節(jié)點(diǎn)。

通過(guò)以上描述可以看到，使用高階API編寫(xiě)分布式TensorFlow應(yīng)用已經(jīng)很方便了，然而因?yàn)镻S架構(gòu)的緣故，我們實(shí)際部署時(shí)，需要規(guī)劃使用多少個(gè)ps，多少個(gè)worker，那么調(diào)試過(guò)程中，需要反復(fù)調(diào)整ps和worker的數(shù)量。當(dāng)模型規(guī)模較大時(shí)，在分布式訓(xùn)練過(guò)程中，ps可能成為網(wǎng)絡(luò)瓶頸，因?yàn)樗衱orker都需要從ps處更新/獲取參數(shù)，如果ps節(jié)點(diǎn)網(wǎng)絡(luò)被打滿，那么worker節(jié)點(diǎn)可能就會(huì)堵塞等待，以至于其計(jì)算能力就發(fā)揮不出來(lái)。所以后面TensorFlow引入All-Reduce架構(gòu)解決這類(lèi)問(wèn)題。

參考

TensorFlow分布式全套

TensorFlow架構(gòu)與設(shè)計(jì)：概述

http://sharkdtu.com/posts/dist-tf-evolution.html

https://zhuanlan.zhihu.com/p/70312627

總結(jié)

以上是生活随笔為你收集整理的[深度学习] 分布式Tensorflow 2.0 介绍（二）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。