spark處理大規模語料庫統計詞匯

最近迷上了spark，寫一個專門處理語料庫生成詞庫的項目拿來練練手， github地址：https://github.com/LiuRoy/spark_splitter。代碼實現參考wordmaker項目，有興趣的可以看一下，此項目用到了不少很tricky的技巧提升性能，單純只想看懂源代碼可以參考wordmaker作者的一份簡單版代碼。

這個項目統計語料庫的結果和執行速度都還不錯，但缺點也很明顯，只能處理GBK編碼的文檔，而且不能分布式運行，剛好最近在接觸spark，所以用python實現了里面的算法，使之能處理更大規模的語料庫，並且同時支持GBK和UTF8兩種編碼格式。

分詞原理

wordmaker提供了一個統計大規模語料庫詞匯的算法，和結巴分詞的原理不同，它不依賴已經統計好的詞庫或者隱馬爾可夫模型，但是同樣能得到不錯的統計結果。原作者的文檔提到是用多個線程獨立計算各個文本塊的詞的信息，再按詞的順序分段合並，再計算各個段的字可能組成詞的概率、左右熵，得到詞語輸出。下面就詳細的講解各個步驟：

1. 讀取文本，去掉文本中的換行、空格、標點，將語料庫分解成一條一條只包含漢字的句子。
2. 將上一步中的所有句子切分成各種長度的詞，並統計所有詞出現的次數。此處的切分只是把所有出現的可能都列出來，舉個例子，天氣真好 就可以切分為：天 天氣 天氣真 天氣真好 氣 氣真 氣真好 真 真好 好。不過為了省略一些沒有必要的計算工作，此處設置了一個詞匯長度限制。
3. 針對上面切分出來的詞匯，為了篩掉出那些多個詞匯連接在一起的情況，會進一步計算這個詞分詞的結果，例如 月亮星星中的月亮和星星這兩個詞匯在步驟二得到的詞匯中頻率非常高，則認為月亮星星不是一個單獨的詞匯，需要被剔除掉。
4. 為了進一步剔除錯誤切分的詞匯，此處用到了信息熵的概念。舉例：邯鄲，邯字右邊只有鄲字一種組合，所以邯的右熵為0，這樣切分就是錯誤的。因為漢字詞語中漢字的關系比較緊密，如果把一個詞切分開來，則他們的熵勢必會小，只需要取一個合適的閾值過濾掉這種錯誤切分即可。

代碼解釋

原始的C++代碼挺長，但是用python改寫之后很少，上文中的1 2 3步用spark實現非常簡單，代碼在split函數中。第3部過濾后的結果已經相對較小，可以直接取出放入內存中，再計算熵過濾，在split中執行target_phrase_rdd.filter(lambda x: _filter(x))過濾的時候可以phrasedictmap做成spark中的廣播變量，提升分布式計算的效率，因為只有一台機器，所以就沒有這樣做。split代碼如下，原來很長的代碼用python很容易就實現了。

def split(self):
    """spark處理"""
    raw_rdd = self.sc.textFile(self.corpus_path)

    utf_rdd = raw_rdd.map(lambda line: str_decode(line))
    hanzi_rdd = utf_rdd.flatMap(lambda line: extract_hanzi(line))

    raw_phrase_rdd = hanzi_rdd.flatMap(lambda sentence: cut_sentence(sentence))

    phrase_rdd = raw_phrase_rdd.reduceByKey(lambda x, y: x + y)
    phrase_dict_map = dict(phrase_rdd.collect())
    total_count = 0
    for _, freq in phrase_dict_map.iteritems():
        total_count += freq

    def _filter(pair):
        phrase, frequency = pair
        max_ff = 0
        for i in xrange(1, len(phrase)):
            left = phrase[:i]
            right = phrase[i:]
            left_f = phrase_dict_map.get(left, 0)
            right_f = phrase_dict_map.get(right, 0)
            max_ff = max(left_f * right_f, max_ff)
        return total_count * frequency / max_ff > 100.0

    target_phrase_rdd = phrase_rdd.filter(lambda x: len(x[0]) >= 2 and x[1] >= 3)
    result_phrase_rdd = target_phrase_rdd.filter(lambda x: _filter(x))
    self.result_phrase_set = set(result_phrase_rdd.keys().collect())
    self.phrase_dict_map = {key: PhraseInfo(val) for key, val in phrase_dict_map.iteritems()}

分詞結果

進入spark_splitter/splitter目錄，執行命令PYTHONPATH=. spark-submit spark.py處理test/moyan.txt文本，是莫言全集，統計完成的結果在out.txt中，統計部分的結果如下，由於算法的原因，帶有 的 地 這種連詞詞語不能正確的切分。

(我也不知道為什么這個詞這么多)

問題統計

1. 上述的算法不能去掉連詞，結果中會出現很多類似於輕輕地 等待着這種詞
2. 單機上用spark處理小規模數據沒有任何優勢，比wordmaker中的C++代碼慢很多
3. 可以和結巴基於隱馬爾科夫模型的分詞結果做一下准確率的比較