更多深度學習在NLP方面應用的經典論文、實踐經驗和最新訊息，歡迎關注微信公眾號“深度學習與NLP” 或掃描頭像二維碼新增關注。

進行深度學習模型訓練的時候，一般使用GPU來進行加速，當訓練樣本只有百萬級別的時候，單卡GPU通常就能滿足我們的需求，但是當訓練樣本量達到上千萬，上億級別之後，單卡訓練耗時很長，這個時候通常需要採用多機多卡加速。深度學習多卡訓練常見有兩種方式，一種是資料並行化（data parallelism），另外一種是模型並行化（model parallelism）。

資料並行化：每個GPU上面跑一個模型，模型與模型之間結構引數相同，只是訓練的資料不一樣，每個模型通過最後的

       資料並行化分為梯度的更新分為同步和非同步兩種模式。

同步模式：等到所有的資料分片都完成了梯度計算並把梯度傳到PS之後統一的更新每個模型的引數。優點是訓練穩定，訓練出來的模型得到精度比較高；缺點是訓練的時間取決於分片中最慢的那個片，所以同步模式適用於GPU之間效能差異不大情況下。

模型並行化：當一個模型非常複雜，非常大，達到單機的記憶體根本沒法容納的時候，模型並行化就是一個好的選擇。直觀說就多多個

 還有資料並行和模型並行的混合模型：

資料並行適用於資料量比較小時快速訓練模型，模型並行適用於大資料、大模型場景下。這裡只是簡單介紹，想要深入瞭解細節可以找其他資料學習一下。下面主要基於tensorflow講一個數據並行化的例項。

1、單機多卡訓練：給個例子，比如一臺機器上裝有4個GPU卡，以cpu做為PS（master），主要儲存一下引數和變數，進行梯度平均。其餘4個GPU訓練模型（worker），進行一些計算量比加大的操作。

    1）本地對資料切分為一個一個batch；

    2）把資料分別放到送到4個GPU進行模型訓練，每個GPU拿到資料不一樣；

    3）四個GPU訓練，求loss得到梯度，並把梯度送回到CPU進行模型平均。

    4）cpu接收4個gpu傳來的梯度，進行梯度平均。

    5）四個GPU跟新引數

    6）重複2~5知道模型收斂。

2、分散式的多久多卡：當是在一個多臺機器的叢集上訓練的時候採用這種方式，在tensorflow中需要明確指定ps和worker的地址，這種方式相容單機多卡，只是把ps和worker的地址設為local就可以了。

下面簡要介紹下tensorflow中支援多卡訓練和引數更新的幾個API，具體介紹可以參考這篇文章（Distributedtensorflow實現原理）http://blog.csdn.net/lqfarmer/article/details/70214026。

Tensorflow進行重複性訓練有In-graph replication和Between-graphreplication兩種方式，In-graph replication就是資料並行化模式，Between-graphreplication就是資料並行化模式。梯度更新有非同步Asynchronous training和同步Synchronous training兩種模式。

Tensorflow官網也給了一個cifar10_multi_gpu_train.py 的例子，在單機多卡上執行，這裡我給一個自己做的單機多卡訓練的簡單例子供參考，自己在搭建這個結構過程中也栽了很多坑，還在繼續探索中，僅有訓練部分。

程式主要分為五個部分：

Main函式：定義主要執行邏輯；

Run_epoch函式：定義主要訓練邏輯；

Generate_feed_dic函式：產生訓練需要的batch樣本；

Multi_gpu_model函式：定義多個tower，每個tower對應一個gpu；

Average_gradients函式：梯度平均計算。

一下是完整程式碼：

#critital class define

#getaverage gradient

defaverage_gradients(tower_grads):

  average_grads = []

  for grad_and_vars in zip(*tower_grads):

    grads = []

    for g, _ in grad_and_vars:

      expanded_g = tf.expand_dims(g, 0)

      grads.append(expanded_g)

    grad = tf.concat(axis=0, values=grads)

    grad = tf.reduce_mean(grad, 0)

    v = grad_and_vars[0][1]

    grad_and_var = (grad, v)

    average_grads.append(grad_and_var)

  return average_grads

#setupmultiple gpu tower            

defmulti_gpu_model(num_gpus=4, word_embeddings = None):

    grads = []

    global_step = tf.Variable(0,name="global_step", trainable=False)

    optimizer = tf.train.AdamOptimizer(1e-3)

    withtf.variable_scope(tf.get_variable_scope()) as initScope:

        for i in range(num_gpus):

            withtf.device("/gpu:%d"%i):

              withtf.name_scope("tower_%d"%i):

                siameseModel = SiameseLSTM(

                                sequence_length=FLAGS.max_document_length,

                               embedding_size=FLAGS.embedding_dim,

                               hidden_units=FLAGS.hidden_units,

                               l2_reg_lambda=FLAGS.l2_reg_lambda,

                                batch_size=FLAGS.batch_size,

                               word_embeddings=word_embeddings)

               tf.get_variable_scope().reuse_variables()

               tf.add_to_collection("train_model", siameseModel)

                grad_and_var =optimizer.compute_gradients(siameseModel.loss)

                grads.append(grad_and_var)

               tf.add_to_collection("loss",siameseModel.loss)

               tf.add_to_collection("accuracy",siameseModel.accuracy)

                tf.add_to_collection("distance",siameseModel.distance)                 

    with tf.device("cpu:0"):

        averaged_gradients =average_gradients(grads)

        train_op =optimizer.apply_gradients(averaged_gradients, global_step=global_step)

    return train_op,global_step

#generating training data      

defgenerate_feed_dic(sess, batch_generator,feed_dict,train_op):

    SMS =tf.get_collection("train_model")

    for siameseModel in SMS:

        x1_batch, x2_batch, y_batch =batch_generator.next()

        if random()>0.5:

            feed_dict[siameseModel.input_x1] =x1_batch

            feed_dict[siameseModel.input_x2] =x2_batch

            feed_dict[siameseModel.input_y] =y_batch

            feed_dict[siameseModel.dropout_keep_prob]= FLAGS.dropout_keep_prob

        else:

            feed_dict[siameseModel.input_x1] =x2_batch

            feed_dict[siameseModel.input_x2] =x1_batch

            feed_dict[siameseModel.input_y] =y_batch

            feed_dict[siameseModel.dropout_keep_prob]= FLAGS.dropout_keep_prob

    return feed_dict           

#define main trainingprocess

def run_epoch(sess,train_x1_idsList,train_x2_idsList,train_y,scope,global_step,train_op=None,is_training=False):

    if is_training:

        epoches = len(train_x1_idsList) //FLAGS.batch_size

        batch_generator =datatool.data_iterator(train_x1_idsList, train_x2_idsList,train_y,FLAGS.batch_size,FLAGS.max_document_length)

#         siameseModels =tf.get_collection("train_model")

        while epoches > 0:

            feed_dict = {}

            epoches -= 1

            feed_dict =generate_feed_dic(sess,batch_generator,feed_dict,train_op)

            i = FLAGS.num_iteration

            while i > 0:

                i = i - 1

                losses =tf.get_collection("loss")

                accuracy =tf.get_collection("accuracy")

                distance =tf.get_collection("distance")

                total_accuracy =tf.add_n(losses, name='total_accu')

                total_distance = tf.add_n(losses,name='total_distance')

                total_loss = tf.add_n(losses,name='total_loss')

                avg_losses = total_loss / 4

                avg_accu = total_accuracy / 4

                avg_dist = total_distance / 4

                time_str =datetime.datetime.now().isoformat()

               _,step,avg_losses,avg_accu,avg_dist =sess.run([train_op,global_step,total_loss,avg_accu,avg_dist],feed_dict)

                                     #輸出訓練精度

                print("TRAIN {}: step {},avg_loss {:g}, avg_dist {:g}, avg_acc {:g}".format(time_str, step,avg_losses, avg_dist, avg_accu))

 #whole training process      

defmain(argv=None):

    print("\nParameters:")

    for attr, value insorted(FLAGS.__flags.items()):

        print("{}={}".format(attr.upper(),value))

    print("")

    #載入詞向量

    word2id, word_embeddings =datatool.load_word2vec("your dir for word2vec")

    print("load train data")

    (train_x1_idsList,train_x2_idsList,train_y),(valid_x1_idsList, valid_x2_lList,valid_y) =datatool.get_data_for_siamese(word2id, FLAGS.data_path)

    print("starting graph def")

    gpu_options =tf.GPUOptions(per_process_gpu_memory_fraction=0.8)

    withtf.Graph().as_default():#,tf.device('/cpu:0')

        session_conf = tf.ConfigProto(

           allow_soft_placement=FLAGS.allow_soft_placement,

           log_device_placement=FLAGS.log_device_placement,

            gpu_options=gpu_options)

        sess = tf.Session(config=session_conf)

        print("started session")

        print ("build multiplemodel")

        with tf.name_scope("train")as train_scope:

            print("define multiple gpumodel and init the training operation")

            train_op,global_step =multi_gpu_model(FLAGS.num_gpus,word_embeddings)

            print ("init allvariable")

           sess.run(tf.global_variables_initializer())

            print ("run epochestage")

           run_epoch(sess,train_x1_idsList,train_x2_idsList,train_y,train_scope,global_step,train_op,True)

        # Checkpoint directory. Tensorflowassumes this directory already exists so we need to create it

        timestamp = str(int(time.time()))

        checkpoint_dir =os.path.abspath(os.path.join(out_dir, "checkpoints"))

        checkpoint_prefix =os.path.join(checkpoint_dir, "model")

        if not os.path.exists(checkpoint_dir):

            os.makedirs(checkpoint_dir)

        out_dir =os.path.abspath(os.path.join(os.path.curdir, "runs", timestamp))

        print("Writing to {}\n".format(out_dir))

        saver =tf.train.Saver(tf.global_variables(), max_to_keep=100)

注意：我用的是已經訓練好的詞向量，這裡只需要載入進來就可以了。