LDA背景

LDA（隱含狄利克雷分佈）是一個主題聚類模型，是當前主題聚類領域最火、最有力的模型之一，它能通過多輪迭代把特徵向量集合按主題分類。目前，廣泛運用在文字主題聚類中。
LDA的開源實現有很多。目前廣泛使用、能夠分散式並行處理大規模語料庫的有微軟的LightLDA，谷歌plda、plda+，sparkLDA等等。下面介紹這3種LDA：

LightLDA依賴於微軟自己實現的multiverso引數伺服器，伺服器底層使用mpi或zeromq傳送訊息。LDA模型(word-topic矩陣)由引數伺服器儲存，它為文件訓練程序提供引數查詢、更新服務。

plda、plda+使用mpi訊息通訊，將mpi程序分為word、doc倆部分。doc程序訓練文件，word程序為doc程序提供模型的查詢、更新功能。

spark LDA有兩種實現：1.基於gibbs sampling原理和使用GraphX實現的版本（即spark文件上所說的EMLDAOptimizer and DistributedLDAModel），2.基於變分推斷原理的實現的版本（即spark文件上的OnlineLDAOptimizer and LocalLDAModel）。Spark LDA的介紹請參見這裡。

LightLDA，plda、plda+，spark LDA比較

論能夠處理預料庫的規模大小，LihgtLDA要遠遠好於plda和spark LDA
經過測試，在10個伺服器（8核40GB）叢集規模下:
LihgtLDA能夠處理上億文件、百萬詞彙的語料庫，能夠訓練上百萬主題數。這樣的處理能力使得LihgtLDA能夠輕鬆訓練絕大多數語料庫。微軟號稱使用幾十機器的叢集便能訓練Bing搜尋引擎爬下資料的十分之一。
相對於LihgtLDA ，plda+能夠處理規模小的多，上限是：詞彙數目*主題數（模型大小） < 5億。當語料庫規模達到上限後，mpi叢集會因記憶體不夠而終止，或因為記憶體資料頻繁切換，迭代速度十分緩慢。雖然plda+對語料庫的詞彙數目和訓練的主題數目很敏感，但對文件的規模並不是很敏感，在詞彙數目和主題數目較小的情況下，1000萬級別的文件也能夠輕鬆解決。
spark LDA的GraphX版處理規模衡量標準是圖的頂點資料，即（文件數 + 詞彙數目）*主題數目，上限是 文件數*主題數 < 50億（由於詞彙數目相對於文件數目往往較小，近似等於 文件數*主題數）。當超過這個規模後，spark叢集進入假死狀態。不停有節點出現OOM，直至任務以失敗告終。
變分推斷實現的spark LDA瓶頸是 詞彙數目*主題數目，這個值也就是我們所說的模型大小，上限約1億。為什麼存在這個瓶頸呢？是因為變分推斷的實現過程中，模型使用矩陣本地儲存，各個分割槽計算模型的部分值，然後在driver上將矩陣reduce疊加。當模型過大，driver節點的記憶體就無法承受各個分割槽發過來的模型。
收斂速度上，LightLDA要遠快於plda、plda+和spark LDA。小規模語料庫（30萬文件，10萬詞，1000主題）測試，LightLDA : plda+ : spark LDA(graphx) = 1:4:50
為什麼各種LDA的能夠處理語料庫規模的衡量標準不一樣呢？這與它們的實現方式有關，不同的LDA有不同的瓶頸，我們這裡單講plda+的原始碼解讀，其他lda後續介紹

plda+介紹

plda+是LDA的並行C++實現，由谷歌公司開發，它分散式基礎是MPI，使用高度優化的Gibbs sampling演算法訓練文件。

如圖所示，plda+將mpi程序組分為2部分——word程序和doc程序。

word程序

word程序儲存plda+的模型，使用分佈是儲存方式，每個程序只負責模型的一部分。LDA的模型指的是word-topic矩陣（矩陣大小=詞彙數目x主題數目），矩陣每行表示語料庫中一word在各個topic中出現的次數。實現上，每行word-topic由向量或陣列表示。word程序負責為doc程序提供word-topic模型引數（即矩陣中的一行，word的各topic出現次數），響應doc程序傳送過來的模型更新訊息。它的角色就相當於一個引數伺服器。

doc程序

doc程序是plda+儲存文件的地方，也是訓練文件的地方。也採用分散式儲存方式，每個程序只持有語料庫的一部分文件。另外，doc程序還分佈是儲存doc-topic矩陣，doc-topic矩陣（矩陣大小=文件數目x主題數目）描述語料庫各文件doc中的所有詞在各個topic下的數目。doc程序從word程序獲取word-topic引數和global_topic引數（每個主題擁有的詞的數目，由word-topic矩陣按行疊加），依據gibbs sampling演算法為每個詞的重新選取主題，將詞的主題選取情況傳送訊息給word程序，通知其更新模型。

doc程序主要由3部分組成：

• 文件集合
  文件由分佈到該doc程序的所有文件組成，各文件記錄了自己的詞頻資訊和自己的各個詞主題選取資訊。
• local word
  local word表示doc程序中所擁有的文件的詞彙集合。程序建立了詞到文件的反轉索引word_inverted_index資料結構，能夠使用word來遍歷所有擁有該詞的文件。
• route（路由表）
  route為每個local word記錄了一個mpi程序號，這個程序號即word程序的編號，表示該local word對應的word-topic模型由這個word程序負責。有了route，doc程序傳送word-topic請求和更新訊息時便知道往哪個word程序傳送了.

MPI訊息

plda+中總是doc程序向word程序主動傳送請訊息，word程序響應doc程序的請求訊息，不存在其他的訊息通訊方式，如doc程序和doc程序之間、word程序和word程序之間，就不存在訊息通訊。

訊息通訊的型別有以下幾種：

• PLDAPLUS_TAG_FETCH
  doc程序向word程序發起word-topic引數請求。訊息通訊的方式是：請求-應答機制，word程序收到請求後向doc程序傳送資料。所有的訊息通訊採用非同步傳送方式，doc程序與word程序傳送訊息後無需等待，繼續做其他的事。
• PLDAPLUS_TAG_FETCH_GLOBAL
  doc程序向word程序請求global_topic引數
• PLDAPLUS_TAG_UPDATE
  doc程序向word程序傳送模型更新訊息，訊息中包含doc程序local word中某個詞的主題變化情況。
• PLDAPLUS_TAG_DONE
  doc程序通知word程序訓練結束，word程序退出訊息等待的主迴圈。這類訊息在最後一輪迭代完畢後，doc程序才傳送。

plda+初始化

plda+要求在叢集各個節點放置一份完整的語料庫文件，各個程序從完整語料庫中抽取文件和詞來初始化一些重要的資料結構。由前面介紹可知，word程序和doc程序所需的資料並不相同，因而他們的初始化行為也不一樣。好在mpi能夠根據程序號來判斷，有區別的讓不同程序執行不同的程式碼。下面是主要的初始化步驟：

所有程序

1. 建立word_index_map
   word_index_map是語料庫詞彙到索引的對映結構(c++ map)，實現上是先將詞彙按字串順序排序，把詞彙對映到序號。之後，使用索引來代表詞彙，由於處理int比string效率要高的多，這個做法可以提升效率。
2. 建立word_pw_map (路由表)
   word_pw_map是local word到word程序的對映，就是我們上述的route結構。

doc程序

1. 將語料庫中的文件進行按doc程序輪流分配，分配完畢後，便確定了各doc程序擁有的文件集合和local word
2. 為文件中的詞隨機選擇主題，形成doc-topic
3. 將local word的主題初始主題情況傳送給word程序，通知其更新模型。

Word程序

1. 按照route路由表為word程序分配word，分配完畢後各程序便擁有各自的local word，進行編號，形成本地索引。建立global_local_word_index_map_，實現語料庫中詞全域性索引到程序中本地索引之間的對映
2. 為本地的詞建立空的word-topic模型，其初始值為0
3. 進行listen，listen是word程序接下來一直進行的事，它在不停地迴圈等待doc程序的訊息，直到接收到所有doc程序的PLDAPLUS_TAG_DONE訊息後才退出
4. listen在初始化階段，word程序主要接收的是doc程序傳送過來的模型更新訊息，形成初始模型。在後續迭代階段，便響應doc程序的各種訊息。

word程序listen實現

就像大多數伺服器程式邏輯一樣，listen不斷執行迴圈，等待訊息，響應訊息…

do {
    MPI_Recv(recv_buf, num_topics_t, MPI_LONG_LONG,
             MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
    int tag = status.MPI_TAG;
    int source = status.MPI_SOURCE;

    switch(tag & 3) {   // get the last two bits
      case PLDAPLUS_TAG_FETCH : {
        MPI_Wait(&req, &status);
        map<int, int>::iterator iter =
            global_local_word_index_map_.find(tag >> PLDAPLUS_TAG_LENGTH);
        if(iter != global_local_word_index_map_.end()) {
          const TopicCountDistribution&   topic_word =  //請求詞的topic引數
              GetWordTopicDistribution(iter->second); //將請求的詞轉為本地索引
          topic_word.replicate(send_buf);
        }
        MPI_Isend(send_buf, num_topics_t, MPI_LONG_LONG,
                  source, tag, MPI_COMM_WORLD, &req); //非同步傳送訊息
        break;
      }
      case PLDAPLUS_TAG_FETCH_GLOBAL : {
                ComputeLocalWordLlh(); //tlz
        if(first_flag) {
          first_flag = false;
        } else {
          MPI_Wait(&req, &status);
        }
        const TopicCountDistribution& global_topic =
            GetGlobalTopicDistribution();
        global_topic.replicate(send_buf);
        MPI_Isend(send_buf, num_topics_t, MPI_LONG_LONG,
                  source, tag, MPI_COMM_WORLD, &req);
        break;
      }
      case PLDAPLUS_TAG_UPDATE : {
        int word_index = global_local_word_index_map_[tag >> PLDAPLUS_TAG_LENGTH];
        for(int k = 0; k < num_topics_t; ++k) {
          IncrementTopic(word_index, k, recv_buf[k]);  //更新模型
        }
        break;
      }
      case PLDAPLUS_TAG_DONE : {
        ++count_done;  //累加doc程序發來的PLDAPLUS_TAG_DONE訊息
        break;
      }
      default : {
        // tag error
      }
    }
  } while(count_done < pdnum);  //收到所有doc程序的PLDAPLUS_TAG_DONE退出listen

word程序就像引數伺服器，不停地為doc程序提供word-topic和global-topic模型引數

doc程序的詞優先順序訓練文件

plda+仍然使用傳統的gibbs sampling演算法，但它在訓練順序上進行了大膽的創新。

原始的文件訓練採用文件優先順序，即為語料庫中每篇文件裡的每一個詞使用gibbs sampling確定新的主題。

plda+為doc程序中的文件建立了詞-文件反轉索引（word_inverted_index），local word中的每個詞能夠索引一系列含有該詞的文件。plda+將訓練過程該為local word中的每一個詞對應的每篇文件進行gibbs sampling，為該文件中該詞選取新的主題。
下面是詞（本地索引是local_word_index）訓練程式碼：

    // Sample for word local_word_index
    for(list<InvertedIndex*>::iterator iter = pldaplus_corpus->word_inverted_index[local_word_index].begin();
        iter != pldaplus_corpus->word_inverted_index[local_word_index].end(); ++iter) {
      SampleNewTopicForWordInDocumentWithDistributions(
          (*iter)->word_index_in_document,
          (*iter)->document_ptr, train_model,
          topic_word, global_topic, delta_topic);
    }

迭代器iter是InvertedIndex結構的指標，它會遍歷指向local_word_index對應的所有InvertedIndex結構，InvertedIndex結構會記錄local_word_index對應的文件（document_ptr），以及該詞在文件中的編號（word_index_in_document）。
gibbs sampling過程會為該文件該詞選擇新的主題，它其實並不複雜，只是依據採用公式通過word-topic（程式碼中是topic_word），global_topic，doc-topic引數，該詞新的主題，並把主題更新資訊記錄在delta_topic陣列中。

plda+按詞優先訓練文件中的詞有以下幾個優勢：
1. 減少了local word主題更新資訊的儲存，當local_word_index對應的所有文件處理完畢，doc程序為該詞傳送模型更新訊息。訓練該詞對應文件的過程只需要一個delta_topic陣列的空間儲存即可。若按文件優先順序來sampling，要將取樣結果更新到模型必然要經歷下列方式之一：每取樣一個詞，就傳送模型更新訊息，這會導致大量的通訊。一篇文件訓練完畢或所有文件訓練完畢才傳送更新模型訊息，這需要記錄所有詞的主題更新資訊，因而會帶來大量儲存開銷。
2. 模型更新速度適宜，文件優先順序中所有文件處理完畢再發送訊息，雖然傳送的更新訊息量非常小，但對模型更新來說，這是一個大同步，會導致模型的收斂速度便慢，甚至會出現抖動。
3. 詞優先訓練文件每處理一個word傳送更新訊息，使得訊息傳送（非同步傳送）與文件訓練交叉執行，使得通訊與計算重疊，提高了系統的吞吐率。

global_topic引數獲取

void PLDAPLUSModelForPd::GetGlobalTopic(int64* global_topic) {
  int num_topics_t = num_topics();
  MPI_Status  status;

  MPI_Send(buf_, 0, MPI_LONG_LONG, 0, PLDAPLUS_TAG_FETCH_GLOBAL, MPI_COMM_WORLD);
  MPI_Recv(global_topic, num_topics_t, MPI_LONG_LONG,
           0, PLDAPLUS_TAG_FETCH_GLOBAL, MPI_COMM_WORLD, &status);
  for(int dest = 1; dest < pwnum_; ++dest) {
    MPI_Send(buf_, 0, MPI_LONG_LONG,
             dest, PLDAPLUS_TAG_FETCH_GLOBAL, MPI_COMM_WORLD);
    MPI_Recv(buf_, num_topics_t, MPI_LONG_LONG,
             dest, PLDAPLUS_TAG_FETCH_GLOBAL, MPI_COMM_WORLD, &status);
    for(int k = 0; k < num_topics_t; ++k) {
      global_topic[k] += buf_[k];
    }
  }
}

doc程序依次向各個word程序傳送PLDAPLUS_TAG_FETCH_GLOBAL訊息，將word程序響應的區域性global_topic累加，形成global_topic。為什麼說是區域性呢？因為global_topic是每個主題擁所有詞的總數目，每個word程序只能統計它自己擁有的那一部分模型。
由於程序數目本來較小，plda+為了實現簡單，doc程序使用同步方式進行通訊。

word-topic引數獲取和訊息的非同步機制

  // Init fetching pool
  for(int i = 0; i < num_words_t && pool_size < PLDAPLUS_MAX_POOL_SIZE; ++i) {
    model_pd_->GetTopicWordNonblocking(i, recv_buf + pool_size * num_topics_t,
                                       request_pool + pool_size);
    word_index_pool[pool_size] = i;
    ++pool_size;
  }

  for(int i = pool_size; i < num_words_t; ++i) {
    // Wait for fetching any topic word distribution
    MPI_Waitany(PLDAPLUS_MAX_POOL_SIZE, request_pool, &request_index, &status);

    // Redirect topic word pointer
    topic_word = recv_buf + request_index * num_topics_t;
    memset(delta_topic, 0, sizeof(*delta_topic) * num_topics_t);

    int local_word_index = word_index_pool[request_index];
        model_pd_->UpdateWordCoverTopic(local_word_index, topic_word);
    // Sample for word local_word_index
    for(list<InvertedIndex*>::iterator iter = pldaplus_corpus->word_inverted_index[local_word_index].begin();
        iter != pldaplus_corpus->word_inverted_index[local_word_index].end(); ++iter) {
      SampleNewTopicForWordInDocumentWithDistributions(
          (*iter)->word_index_in_document,
          (*iter)->document_ptr, train_model,
          topic_word, global_topic, delta_topic);
    }

    // Update for word local_word_index
    model_pd_->UpdateTopicWord(local_word_index, delta_topic);
    for(int k = 0; k < num_topics_t; ++k) {
      global_topic[k] += delta_topic[k];
    }

    // Fetch next topic word distribution
    model_pd_->GetTopicWordNonblocking(i, topic_word, request_pool + request_index);
    word_index_pool[request_index] = i;
  }

doc程序使用非同步的MPI訊息請求word_topic引數，非同步方式即請求後不原地等待word程序的響應。plda+的實現如下：
1. plda+在為local word請求word-topic引數時，最開始發出一池子（100個）的word_topic請求，將他們放到訊息池中，監控池子中響應的到來。
2. 每到來一個響應，便根據響應訊息帶來的word_topic引數訓練該詞對應的一系列文件。訓練完畢後，傳送該詞的模型更新訊息。該詞處理完畢，在訊息池中佔的位置也可被佔用。
3. 傳送下一個local word的word-topic引數請求，用前一個詞在訊息池中的位置來存放請求的訊息，進行監控。直到local word全部訓練完畢。

plda+ loglikelihood計算問題

使用過plda+的同學可能發現，plda+的loglikelihood的值竟然是隨迭代次數增加而遞減的，這嚴重不符合likelihood的定義，隨著迭代加深，似然函式的值應該不斷逼近最大值。
筆者依照LihgtLda的計算方式重新實現了plda+的loglikelihood計算。參見：https://github.com/tanglizhe1105/plda
我們把loglikelihood的計算拆成2部分：doc-topic矩陣和word-topic模型矩陣，其中word-topic為了方便計算又分為了word loglikelihood和normalized loglikelihood。拆分的理由請見：https://github.com/Microsoft/lightlda/issues/9
因而，word-topic的loglikelihood為 word +normalized loglikelihood。
總的loglikelihood = doc + word + normalized loglikelihood。

作者介紹

唐黎哲，國防科學技術大學 並行與分散式計算國家重點實驗室(PDL)碩士，從事spark、圖計算、LDA（主題分類）研究，歡迎交流，請多指教。
郵箱：[email protected]