深度學習利器：分布式TensorFlow及實例分析
TensorFlow發展及使用簡介
2015年11月9日谷歌開源了人工智能系統TensorFlow，同時成為2015年最受關注的開源項目之一。TensorFlow的開源大大降低了深度學習在各個行業中的應用難度。TensorFlow的近期里程碑事件主要有：
2016年11月09日：TensorFlow開源一周年。
2016年09月27日：TensorFlow支持機器翻譯模型。
2016年08月30日：TensorFlow支持使用TF-Slim接口定義復雜模型。
2016年08月24日：TensorFlow支持自動學習生成文章摘要模型。
2016年06月29日：TensorFlow支持Wide & Deep Learning。
2016年06月27日：TensorFlow v0.9發布，改進了移動設備的支持。
2016年05月12日：發布SyntaxNet，最精確的自然語言處理模型。
2016年04月29日：DeepMind模型遷移到TensorFlow。
2016年04月14日：發布了分布式TensorFlow。
TensorFlow是一種基于圖計算的開源軟件庫，圖中節點表示數學運算，圖中的邊表示多維數組（Tensor）。TensorFlow是跨平臺的深度學習框架，支持CPU和GPU的運算，支持臺式機、服務器、移動平臺的計算，并從r0.12版本開始支持Windows平臺。Tensorflow提供了各種安裝方式，包括Pip安裝，Virtualenv安裝，Anaconda安裝，docker安裝，源代碼安裝。 本文主要介紹Pip的安裝方式，Pip是一個Python的包安裝及管理工具。Linux系統下，使用Pip的安裝流程如下：
yum install python-pip python-dev
export TF_BINARY_URL=https://storage.googleapis.com/tensorflow/linux/cpu/tensorflow-0.12.0rc0-cp27-none-linux_x86_64.whl
pip install --upgrade $TF_BINARY_URL</code></pre>
安裝完畢后，TensorFlow會安裝到/usr/lib/python2.7/site-packages/tensorflow目錄下。使用TensorFlow之前，我們需要先熟悉下常用API。
tf.random_uniform([1], -1.0, 1.0)：構建一個tensor, 該tensor的shape為[1]，該值符合[-1, 1)的均勻分布。其中[1]表示一維數組，里面包含1個元素。
tf.Variable(initial_value=None)：構建一個新變量，該變量會加入到TensorFlow框架中的圖集合中。
tf.zeros([1])：構建一個tensor, 該tensor的shape為[1], 里面所有元素為0。
tf.square(x, name=None)：計算tensor的平方值。
tf.reduce_mean(input_tensor)：計算input_tensor中所有元素的均值。
tf.train.GradientDescentOptimizer(0.5)：構建一個梯度下降優化器，0.5為學習速率。學習率決定我們邁向（局部）最小值時每一步的步長，設置的太小，那么下降速度會很慢，設的太大可能出現直接越過最小值的現象。所以一般調到目標函數的值在減小而且速度適中的情況。
optimizer.minimize(loss)：構建一個優化算子操作。使用梯度下降法計算損失方程的最小值。loss為需要被優化的損失方程。
tf.initialize_all_variables()：初始化所有TensorFlow的變量。
tf.Session()：創建一個TensorFlow的session，在該session種會運行TensorFlow的圖計算模型。
sess.run()：在session中執行圖模型的運算操作。如果參數為tensor時，可以用來求tensor的值。
下面為使用TensorFlow中的梯度下降法構建線性學習模型的使用示例：
#導入TensorFlow python API庫
import tensorflow as tf
import numpy as np
隨機生成100點（x，y）
x_data = np.random.rand(100).astype(np.float32)
y_data = x_data * 0.1 + 0.3
構建線性模型的tensor變量W, b
W = tf.Variable(tf.random_uniform([1], -1.0, 1.0))
b = tf.Variable(tf.zeros([1]))
y = W * x_data + b
構建損失方程，優化器及訓練模型操作train
loss = tf.reduce_mean(tf.square(y - y_data))
optimizer = tf.train.GradientDescentOptimizer(0.5)
train = optimizer.minimize(loss)
構建變量初始化操作init
init = tf.initialize_all_variables()
構建TensorFlow session
sess = tf.Session()
初始化所有TensorFlow變量
sess.run(init)
訓練該線性模型，每隔20次迭代，輸出模型參數
for step in range(201):
sess.run(train)
if step % 20 == 0:
print(step, sess.run(W), sess.run(b))</code></pre>
分布式TensorFlow應用架構
2016年4月14日，Google發布了分布式TensorFlow，能夠支持在幾百臺機器上并行訓練。分布式的TensorFlow由高性能的gRPC庫作為底層技術支持。TensorFlow集群由一系列的任務組成，這些任務執行TensorFlow的圖計算。每個任務會關聯到TensorFlow的一個服務，該服務用于創建TensorFlow會話及執行圖計算。TensorFlow集群也可以劃分為一個或多個作業，每個作業可以包含一個或多個任務。在一個TensorFlow集群中，通常一個任務運行在一個機器上。如果該機器支持多GPU設備，可以在該機器上運行多個任務，由應用程序控制任務在哪個GPU設備上運行。
常用的深度學習訓練模型為數據并行化，即TensorFlow任務采用相同的訓練模型在不同的小批量數據集上進行訓練，然后在參數服務器上更新模型的共享參數。TensorFlow支持同步訓練和異步訓練兩種模型訓練方式。
異步訓練即TensorFlow上每個節點上的任務為獨立訓練方式，不需要執行協調操作，如下圖所示：
同步訓練為TensorFlow上每個節點上的任務需要讀入共享參數，執行并行化的梯度計算，然后將所有共享參數進行合并，如下圖所示：
分布式TensorFlow 應用開發API主要包括：
tf.train.ClusterSpec({"ps": ps_hosts, "worker": worker_hosts})： 創建TensorFlow集群描述信息，其中ps，worker為作業名稱，ps_hosts，worker_hosts為該作業的任務所在節點的地址信息。示例如下：
cluster = tf.train.ClusterSpec({"worker": ["worker0.example.com:2222", "worker1.example.com:2222", "worker2.example.com:2222"], "ps": ["ps0.example.com:2222", "ps1.example.com:2222"]})
tf.train.Server(cluster, job_name, task_index)：創建一個TensorFlow服務，用于運行相應作業上的計算任務，運行的任務在task_index指定的機器上啟動。
tf.device(device_name_or_function)：設定在指定的設備上執行Tensor運算，示例如下：
#指定在task0所在的機器上執行Tensor的操作運算
with tf.device("/job:ps/task:0"):
weights_1 = tf.Variable(...)
biases_1 = tf.Variable(...)
分布式TensorFlow MNIST模型訓練
MNIST是一個手寫數字的圖片數據庫，可從網站 http://yann.lecun.com/exdb/mnist/ 下載相關數據，其中的每一張圖片為0到9之間的手寫數字灰度圖片，大小為28*28像素，如下圖所示：
MNIST數據集主要包含訓練樣本60000個，測試樣本10000個。圖像數據主要為圖片的像素數據，圖像數據標簽主要表示該圖片的類別。由以下四個文件組成：
train-images-idx3-ubyte.gz （訓練圖像數據60000個）
train-labels-idx1-ubyte.gz （訓練圖像數據標簽60000個）
t10k-images-idx3-ubyte.gz （測試圖像數據10000個）
t10k-labels-idx1-ubyte.gz （測試圖像數據標簽10000個）
本文采用如下的結構對MNIST數據集進行分布式訓練，由三個節點組成。ww01節點為Parameter Server，ww02節點為Worker0，ww03節點為Worker1。其中Parameter Server執行參數更新任務，Worker0，Worker1執行圖模型訓練計算任務，如下圖所示。分布式MNIST訓練模型在執行十萬次迭代后，收斂精度達到97.77%。
在ww01節點執行如下命令，啟動參數服務/job:ps/task:0：
python asyncmnist.py --ps_hosts=ww01:2222 --worker_hosts=ww02:2222,ww03:2222 --job_name=ps --task_index=0
在ww02節點執行如下命令，啟動模型運算/job:worker/task:0：
python asyncmnist.py --ps_hosts=ww01:2222 --worker_hosts=ww02:2222,ww03:2222 --job_name=worker --task_index=0
在ww03節點執行如下命令，啟動模型運算/job:worker/task:1：
python asyncmnist.py --ps_hosts=ww01:2222 --worker_hosts=ww02:2222,ww03:2222 --job_name=worker --task_index=1
分布式MNIST的訓練模型如下：
import math
import tensorflow as tf
from tensorflow.examples.tutorials.mnist import input_data
# TensorFlow集群描述信息，ps_hosts表示參數服務節點信息，worker_hosts表示worker節點信息
tf.app.flags.DEFINE_string("ps_hosts", "", "Comma-separated list of hostname:port pairs")
tf.app.flags.DEFINE_string("worker_hosts", "", "Comma-separated list of hostname:port pairs")
# TensorFlow Server模型描述信息，包括作業名稱，任務編號，隱含層神經元數量，MNIST數據目錄以及每次訓練數據大小（默認一個批次為100個圖片）
tf.app.flags.DEFINE_string("job_name", "", "One of 'ps', 'worker'")
tf.app.flags.DEFINE_integer("task_index", 0, "Index of task within the job")
tf.app.flags.DEFINE_integer("hidden_units", 100, "Number of units in the hidden layer of the NN")
tf.app.flags.DEFINE_string("data_dir", "MNIST_data", "Directory for storing mnist data")
tf.app.flags.DEFINE_integer("batch_size", 100, "Training batch size")
FLAGS = tf.app.flags.FLAGS
#圖片像素大小為28*28像素
IMAGE_PIXELS = 28
def main(_):
#從命令行參數中讀取TensorFlow集群描述信息
ps_hosts = FLAGS.ps_hosts.split(",")
worker_hosts = FLAGS.worker_hosts.split(",")
# 創建TensorFlow集群描述對象
cluster = tf.train.ClusterSpec({"ps": ps_hosts, "worker": worker_hosts})
# 為本地執行Task，創建TensorFlow本地Server對象.
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":
#分配操作到指定的worker上執行，默認為該節點上的cpu0
with tf.device(tf.train.replica_device_setter(worker_device="/job:worker/task:%d" % FLAGS.task_index, cluster=cluster)):
# 定義TensorFlow隱含層參數變量，為全連接神經網絡隱含層
hid_w = tf.Variable(tf.truncated_normal([IMAGE_PIXELS * IMAGE_PIXELS, FLAGS.hidden_units], stddev=1.0 / IMAGE_PIXELS), name="hid_w")
hid_b = tf.Variable(tf.zeros([FLAGS.hidden_units]), name="hid_b")
# 定義TensorFlow softmax回歸層的參數變量
sm_w = tf.Variable(tf.truncated_normal([FLAGS.hidden_units, 10], tddev=1.0 / math.sqrt(FLAGS.hidden_units)), name="sm_w")
sm_b = tf.Variable(tf.zeros([10]), name="sm_b")
#定義模型輸入數據變量（x為圖片像素數據，y_為手寫數字分類）
x = tf.placeholder(tf.float32, [None, IMAGE_PIXELS * IMAGE_PIXELS])
y_ = tf.placeholder(tf.float32, [None, 10])
#定義隱含層及神經元計算模型
hid_lin = tf.nn.xw_plus_b(x, hid_w, hid_b)
hid = tf.nn.relu(hid_lin)
#定義softmax回歸模型，及損失方程
y = tf.nn.softmax(tf.nn.xw_plus_b(hid, sm_w, sm_b))
loss = -tf.reduce_sum(y_ * tf.log(tf.clip_by_value(y, 1e-10, 1.0)))
#定義全局步長，默認值為0
global_step = tf.Variable(0)
#定義訓練模型，采用Adagrad梯度下降法
train_op = tf.train.AdagradOptimizer(0.01).minimize(loss, global_step=global_step)
#定義模型精確度驗證模型，統計模型精確度
correct_prediction = tf.equal(tf.argmax(y,1), tf.argmax(y_,1))
accuracy = tf.reduce_mean(tf.cast(correct_prediction, tf.float32))
#對模型定期做checkpoint，通常用于模型回復
saver = tf.train.Saver()
#定義收集模型統計信息的操作
summary_op = tf.merge_all_summaries()
#定義操作初始化所有模型變量
init_op = tf.initialize_all_variables()
#創建一個監管程序，用于構建模型檢查點以及計算模型統計信息。
sv = tf.train.Supervisor(is_chief=(FLAGS.task_index == 0), logdir="/tmp/train_logs", init_op=init_op, summary_op=summary_op, saver=saver, global_step=global_step, save_model_secs=600)
#讀入MNIST訓練數據集
mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True)
#創建TensorFlow session對象，用于執行TensorFlow圖計算
with sv.managed_session(server.target) as sess:
step = 0
while not sv.should_stop() and step < 1000:
# 讀入MNIST的訓練數據，默認每批次為100個圖片
batch_xs, batch_ys = mnist.train.next_batch(FLAGS.batch_size)
train_feed = {x: batch_xs, y_: batch_ys}
#執行分布式TensorFlow模型訓練
_, step = sess.run([train_op, global_step], feed_dict=train_feed)
#每隔100步長，驗證模型精度
if step % 100 == 0:
print "Done step %d" % step
print(sess.run(accuracy, feed_dict={x: mnist.test.images, y_: mnist.test.labels}))
# 停止TensorFlow Session
sv.stop()
if __name__ == "__main__":
tf.app.run()
梯度下降法在分布式TensorFlow中的性能比較分析
2016年谷歌在ICLR（the International Conference on Learning Representations） Workshop上發表了論文REVISITING DISTRIBUTED SYNCHRONOUS SGD。基于ImageNet數據集，該論文對異步隨機梯度下降法（Async-SGD）和同步隨機梯度下降法（Sync-SGD）進行了比較分析。
Dean在2012年提出了分布式隨機梯度下降法，模型參數可以分布式地存儲在不同的服務器上，稱之為參數服務器（Parameter Server，PS），以及Worker節點可以并發地處理訓練數據并且能夠和參數服務通信獲取模型參數。異步隨機梯度下降法（Async-SGD），主要由以下幾個步驟組成：
- 針對當前批次的訓練數據，從參數服務器獲取模型的最新參數。
- 基于上述獲取到的模型參數，計算損失方程的梯度。
- 將上述計算得到的梯度發送回參數服務器，并相應地更新模型參數。
同步隨機梯度下降法（Sync-SGD）與Sync-SGD的主要差異在于參數服務器將等待所有Worker發送相應的梯度值，并聚合這些梯度值，最后把更新后的梯度值發送回節點。
Async-SGD 的主要問題是每個Worker節點計算的梯度值發送回參數服務器會有參數更新沖突，一定程度影響算法的收斂速度。Sync-SGD算法能夠保證在數據集上執行的是真正的隨機梯度下降法，消除掉了參數的更新沖突。但同步隨機梯度下降法同時帶來的問題是訓練數據的批量數據會比較大，參數服務器上參數的更新時間依賴于最慢的worker節點。
為了解決有些worker節點比較慢的問題，我們可以使用多一點的Worker節點，這樣Worker節點數變為N+N*5%，N為集群Worker節點數。Sync-SGD可以設定為在接受到N個Worker節點的參數后，可以直接更新參數服務器上的模型參數，進入下一個批次的模型訓練。慢節點上訓練出來的參數是會被丟棄掉。我們稱這種方法為Sync-SGD with backups。
2015年，Abadi使用TensorFlow的Async-SGD, Sync-SGD，Sync-SGD with backups訓練模型對ImageNet的Benchmark問題進行了實驗分析。要對該訓練數據進行1000種圖片的分類訓練，實驗環境為50到200個的worker節點，每個worker節點上運行k40 GPU。使用分布式TensorFlow后大大縮短了模型訓練時間，Async-SGD算法實驗結果如下，其中200個節點的訓練時間比采用25個節點的運算時間縮短了8倍，如下圖所示。
下圖為50個Worker節點的Async-SGD, Sync-SGD，Sync-SGD with backups模型訓練結果的比較。
從結果中可以看出增加2個backup節點，Sync-SGD with backups模型可以快速提升模型訓練速度。同時Sync-SGD模型比Async-SGD模型大概提升了25%的訓練速度，以及0.48%的精確度。隨著數據集的增加，分布式訓練的架構變得越來越重要。而分布式TensorFlow正是解決該問題的利器，有效地提升了大規模模型訓練的效率，提供了企業級的深度學習解決方案。
來自：http://www.infoq.com/cn/articles/tensorflow-IBM-casestudy-wuwei