Alink漫談(六) : TF-IDF算法的實現

目錄

• Alink漫談(六) : TF-IDF算法的實現
  • 0x00 摘要
  • 0x01 TF-IDF
    • 1.1 原理
    • 1.2 計算方法
  • 0x02 Alink示例代碼
    • 2.1 示例代碼
    • 2.2 TF-IDF模型
    • 2.3 TF-IDF預測
  • 0x03 分詞 Segment
    • 3.1 結巴分詞
    • 3.2 分詞過程
  • 0x04 訓練
    • 4.1 計算IDF
    • 4.2 排序
      • 4.2.1 SortUtils.pSort
        • 採樣SampleSplitPoint
        • 歸併 SplitPointReducer
        • SplitData把真實數據IDF插入
        • reduceGroup計算同類型單詞數目
      • 4.2.2 localSort
    • 4.3 過濾
  • 0x05 生成模型
    • 5.1 DocCountVectorizerModelData
    • 5.2 BuildDocCountModel
  • 0x06 預測
  • 0x07 參考

0x00 摘要

Alink 是阿里巴巴基於實時計算引擎 Flink 研發的新一代機器學習算法平台，是業界首個同時支持批式算法、流式算法的機器學習平台。TF-IDF（term frequency–inverse document frequency）是一種用於信息檢索與數據挖掘的常用加權技術。本文將為大家展現Alink如何實現TF-IDF。

0x01 TF-IDF

TF-IDF（term frequency–inverse document frequency）是一種統計方法，一種用於信息檢索與數據挖掘的常用加權技術。

TF是詞頻(Term Frequency)，IDF是逆文本頻率指數(Inverse Document Frequency)。

為什麼要用TF-IDF？因為計算機只能識別数字，對於一個一個的單詞，計算機是看不懂的，更別說是一句話，或是一篇文章。而TF-IDF就是用來將文本轉換成計算機看得懂的語言，或者說是機器學習或深度學習模型能夠進行學習訓練的數據集。

1.1 原理

TF-IDF用以評估一個詞對於一個文件集或一個語料庫中的其中一份文件的重要程度。字詞的重要性隨着它在文件中出現的次數成正比增加，但同時會隨着它在語料庫中出現的頻率成反比下降。

TF-IDF的主要思想是：如果某個詞或短語在一篇文章中出現的頻率TF高，並且在其他文章中很少出現，則認為此詞或者短語具有很好的類別區分能力，適合用來分類。

TF-IDF實際上是：TF * IDF，TF詞頻(Term Frequency)，IDF逆向文件頻率(Inverse Document Frequency)。

詞頻（term frequency，TF）指的是某一個給定的詞語在該文件中出現的頻率。這個数字是對詞數(term count)的歸一化，以防止它偏向長的文件（同一個詞語在長文件里可能會比短文件有更高的詞數，而不管該詞語重要與否）。

而IDF逆向文件頻率 (inverse document frequency, IDF)反應了一個詞在所有文本（整個文檔）中出現的頻率，如果一個詞在很多的文本中出現，那麼它的IDF值應該低。而反過來如果一個詞在比較少的文本中出現，那麼它的IDF值應該高。比如一些專業的名詞如“Machine Learning”。這樣的詞IDF值應該高。一個極端的情況，如果一個詞在所有的文本中都出現，那麼它的IDF值應該為0。

如果單單以TF或者IDF來計算一個詞的重要程度都是片面的，因此TF-IDF綜合了TF和IDF兩者的優點，用以評估一字詞對於一個文件集或一個語料庫中的其中一份文件的重要程度。字詞的重要性隨着它在文件中出現的次數成正比增加，但同時會隨着它在語料庫中出現的頻率成反比下降。上述引用總結就是：一個詞語在一篇文章中出現次數越多, 同時在所有文檔中出現次數越少, 越能夠代表該文章，越能與其它文章區分開來。

1.2 計算方法

TF的計算公式如下：

\[TF_w = \frac {N_w}{N} \]

其中 N_w 是在某一文本中詞條w出現的次數，N 是該文本總詞條數。

IDF的計算公式如下：

\[IDF_w = log(\frac {Y}{Y_w + 1}) \]

其中 Y 是語料庫的文檔總數，Y_w 是包含詞條w的文檔數，分母加一是為了避免w 未出現在任何文檔中從而導致分母為0 的情況。

TF-IDF 就是將TF和IDF相乘 :

\[TF-IDF_w = TF_w * IDF_w \]

從以上計算公式便可以看出，某一特定文件內的高詞語頻率，以及該詞語在整個文件集合中的低文件頻率，可以產生出高權重的TF-IDF。因此，TF-IDF傾向於過濾掉常見的詞語，保留重要的詞語。

0x02 Alink示例代碼

2.1 示例代碼

首先我們給出示例代碼，下文是通過一些語料來訓練出一個模型，然後用這個模型來做預測：

public class DocCountVectorizerExample {

    AlgoOperator getData(boolean isBatch) {
        Row[] rows = new Row[]{
                Row.of(0, "二手舊書:醫學電磁成像"),
                Row.of(1, "二手美國文學選讀（ 下冊 ）李宜燮南開大學出版社 9787310003969"),
                Row.of(2, "二手正版圖解象棋入門/謝恩思主編/華齡出版社"),
                Row.of(3, "二手中國糖尿病文獻索引"),
                Row.of(4, "二手郁達夫文集（ 國內版 ）全十二冊館藏書")
        };

        String[] schema = new String[]{"id", "text"};

        if (isBatch) {
            return new MemSourceBatchOp(rows, schema);
        } else {
            return new MemSourceStreamOp(rows, schema);
        }
    }

    public static void main(String[] args) throws Exception {
        DocCountVectorizerExample test = new DocCountVectorizerExample();
        BatchOperator batchData = (BatchOperator) test.getData(true);

         // 分詞
        SegmentBatchOp segment = new SegmentBatchOp() 
                                                .setSelectedCol("text")
                                                .linkFrom(batchData);
        // TF-IDF訓練
        DocCountVectorizerTrainBatchOp model = new DocCountVectorizerTrainBatchOp()
                                                .setSelectedCol("text")
                                                .linkFrom(segment);
        // TF-IDF預測
        DocCountVectorizerPredictBatchOp predictBatch = new 
            																		DocCountVectorizerPredictBatchOp()
                                                .setSelectedCol("text")
                                                .linkFrom(model, segment);
        model.print();
        predictBatch.print();
    }
}

2.2 TF-IDF模型

TF-IDF模型打印出來如下：

model_id|model_info
--------|----------
0|{"minTF":"1.0","featureType":"\"WORD_COUNT\""}
1048576|{"f0":"二手","f1":0.0,"f2":0}
2097152|{"f0":"/","f1":1.0986122886681098,"f2":1}
3145728|{"f0":"出版社","f1":0.6931471805599453,"f2":2}
4194304|{"f0":"）","f1":0.6931471805599453,"f2":3}
5242880|{"f0":"（","f1":0.6931471805599453,"f2":4}
6291456|{"f0":"入門","f1":1.0986122886681098,"f2":5}
......
36700160|{"f0":"美國","f1":1.0986122886681098,"f2":34}
37748736|{"f0":"謝恩","f1":1.0986122886681098,"f2":35}
38797312|{"f0":"象棋","f1":1.0986122886681098,"f2":36}

2.3 TF-IDF預測

TF-IDF預測結果如下：

id|text
--|----
0|$37$0:1.0 6:1.0 10:1.0 25:1.0 26:1.0 28:1.0
1|$37$0:1.0 1:1.0 2:1.0 4:1.0 11:1.0 15:1.0 16:1.0 19:1.0 20:1.0 32:1.0 34:1.0
2|$37$0:1.0 3:2.0 4:1.0 5:1.0 8:1.0 22:1.0 23:1.0 24:1.0 29:1.0 35:1.0 36:1.0
3|$37$0:1.0 12:1.0 27:1.0 31:1.0 33:1.0
4|$37$0:1.0 1:1.0 2:1.0 7:1.0 9:1.0 13:1.0 14:1.0 17:1.0 18:1.0 21:1.0 30:1.0

0x03 分詞 Segment

中文分詞(Chinese Word Segmentation) 指的是將一個漢字序列切分成一個一個單獨的詞。分詞就是將連續的字序列按照一定的規範重新組合成詞序列的過程。

示例代碼中，分詞部分如下：

    SegmentBatchOp segment = new SegmentBatchOp() 
                                            .setSelectedCol("text")
                                            .linkFrom(batchData);

分詞主要是如下兩個類，其作用就是把中文文檔分割成單詞。

public final class SegmentBatchOp extends MapBatchOp <SegmentBatchOp>
	implements SegmentParams <SegmentBatchOp> {

	public SegmentBatchOp(Params params) {
		super(SegmentMapper::new, params);
	}
}

public class SegmentMapper extends SISOMapper {
	private JiebaSegmenter segmentor;
}

3.1 結巴分詞

有經驗的同學看到這裏就會露出微笑：結巴分詞。

jieba分詞是國內使用人數最多的中文分詞工具https://github.com/fxsjy/jieba。jieba分詞支持四種分詞模式：

• 精確模式，試圖將句子最精確地切開，適合文本分析；
• 全模式，把句子中所有的可以成詞的詞語都掃描出來, 速度非常快，但是不能解決歧義；
• 搜索引擎模式，在精確模式的基礎上，對長詞再次切分，提高召回率，適合用於搜索引擎分詞。
• paddle模式，利用PaddlePaddle深度學習框架，訓練序列標註（雙向GRU）網絡模型實現分詞。

Alink使用了com.alibaba.alink.operator.common.nlp.jiebasegment.viterbi.FinalSeg;來 完成分詞。具體是在https://github.com/huaban/jieba-analysis的基礎上稍微做了調整。

public class JiebaSegmenter implements Serializable {
    private static FinalSeg finalSeg = FinalSeg.getInstance();
    private WordDictionary wordDict;
    ......
    private Map<Integer, List<Integer>> createDAG(String sentence) 
}

從Alink代碼中看，實現了索引分詞和查詢分詞兩種模式，應該是有分詞粒度粗細之分。

createDAG函數的作用是 ：在處理句子過程中，基於前綴詞典實現高效的詞圖掃描，生成句子中漢字所有可能成詞情況所構成的有向無環圖 (DAG)。

結巴分詞對於未登錄詞，採用了基於漢字成詞能力的 HMM 模型，使用了 Viterbi 算法。

3.2 分詞過程

分詞過程主要是在SegmentMapper.mapColumn函數中完成的，當輸入是 “二手舊書:醫學電磁成像”，結巴分詞將這個句子分成了六個單詞。具體參見如下：

input = "二手舊書:醫學電磁成像"
tokens = {ArrayList@9619}  size = 6
 0 = {SegToken@9630} "[二手, 0, 2]"
 1 = {SegToken@9631} "[舊書, 2, 4]"
 2 = {SegToken@9632} "[:, 4, 5]"
 3 = {SegToken@9633} "[醫學, 5, 7]"
 4 = {SegToken@9634} "[電磁, 7, 9]"
 5 = {SegToken@9635} "[成像, 9, 11]"
 
mapColumn:44, SegmentMapper (com.alibaba.alink.operator.common.nlp)
apply:-1, 35206803 (com.alibaba.alink.common.mapper.SISOMapper$$Lambda$646)
handleMap:75, SISOColsHelper (com.alibaba.alink.common.mapper)
map:52, SISOMapper (com.alibaba.alink.common.mapper)
map:21, MapperAdapter (com.alibaba.alink.common.mapper)
map:11, MapperAdapter (com.alibaba.alink.common.mapper)
collect:79, ChainedMapDriver (org.apache.flink.runtime.operators.chaining)
collect:35, CountingCollector (org.apache.flink.runtime.operators.util.metrics)
invoke:196, DataSourceTask (org.apache.flink.runtime.operators)

0x04 訓練

訓練是在DocCountVectorizerTrainBatchOp類完成的，其通過linkFrom完成了模型的構建。其實計算TF IDF相對 簡單，複雜之處在於之後的大規模排序。

public DocCountVectorizerTrainBatchOp linkFrom(BatchOperator<?>... inputs) {
        BatchOperator<?> in = checkAndGetFirst(inputs);
  
        DataSet<DocCountVectorizerModelData> resDocCountModel = generateDocCountModel(getParams(), in);

        DataSet<Row> res = resDocCountModel.mapPartition(new MapPartitionFunction<DocCountVectorizerModelData, Row>() {
            @Override
            public void mapPartition(Iterable<DocCountVectorizerModelData> modelDataList, Collector<Row> collector) {
                new DocCountVectorizerModelDataConverter().save(modelDataList.iterator().next(), collector);
            }
        });
        this.setOutput(res, new DocCountVectorizerModelDataConverter().getModelSchema());
        return this;
}

4.1 計算IDF

計算 IDF 的工作是在generateDocCountModel完成的，具體步驟如下：

第一步 通過DocWordSplitCount和UDTF的混合使用得到了文檔中的單詞數目docWordCnt。

BatchOperator<?> docWordCnt = in.udtf(
        params.get(SELECTED_COL),
        new String[] {WORD_COL_NAME, DOC_WORD_COUNT_COL_NAME},
        new DocWordSplitCount(NLPConstant.WORD_DELIMITER),
        new String[] {});

DocWordSplitCount.eval的輸入是已經分詞的句子，然後按照空格分詞，按照單詞計數。其結果是：

map = {HashMap@9816}  size = 6
 "醫學" -> {Long@9833} 1
 "電磁" -> {Long@9833} 1
 ":" -> {Long@9833} 1
 "成像" -> {Long@9833} 1
 "舊書" -> {Long@9833} 1
 "二手" -> {Long@9833} 1

第二步 得到了文檔數目docCnt

BatchOperator docCnt = in.select("COUNT(1) AS " + DOC_COUNT_COL_NAME);

這個數目會廣播出去 .withBroadcastSet(docCnt.getDataSet(), "docCnt");，後面的CalcIdf會繼續使用，進行 行數統計。

第三步 會通過CalcIdf計算出每一個單詞的DF和IDF。

open時候會獲取docCnt。然後reduce會計算IDF，具體計算如下：

double idf = Math.log((1.0 + docCnt) / (1.0 + df));
collector.collect(Row.of(featureName, -wordCount, idf));

具體得到如下

df = 1.0
wordCount = 1.0
featureName = "中國"
idf = 1.0986122886681098
docCnt = 5

這裏一個重點是：返回值中，是 -wordCount，因為單詞越多權重越小，為了比較所以取負。

4.2 排序

得到所有單詞的IDF之後，就得到了一個IDF字典，這時候需要對字典按照權重進行排序。排序具體分為兩步。

4.2.1 SortUtils.pSort

第一步是SortUtils.pSort，大規模并行抽樣排序。

Tuple2<DataSet<Tuple2<Integer, Row>>, DataSet<Tuple2<Integer, Long>>> partitioned = SortUtils.pSort(sortInput, 1);

這步非常複雜，Alink參考了論文，如果有興趣的兄弟可以深入了解下。

* reference: Yang, X. (2014). Chong gou da shu ju tong ji (1st ed., pp. 25-29).
* Note: This algorithm is improved on the base of the parallel sorting by regular sampling(PSRS).

pSort返回值是：

* @return f0: dataset which is indexed by partition id, f1: dataset which has partition id and count.

pSort中又分如下幾步

採樣SampleSplitPoint

SortUtils.SampleSplitPoint.mapPartition這裏完成了採樣。

DataSet <Tuple2 <Object, Integer>> splitPoints = input
   .mapPartition(new SampleSplitPoint(index))
   .reduceGroup(new SplitPointReducer());

這裏的輸入row就是上文IDF的返回數值。

用allValues記錄了本task目前處理的句子有多少個單詞。

用splitPoints做了採樣。如何選擇呢，通過genSampleIndex函數。

public static Long genSampleIndex(Long splitPointIdx, Long count, Long splitPointSize) {
   splitPointIdx++;
   splitPointSize++;

   Long div = count / splitPointSize;
   Long mod = count % splitPointSize;

   return div * splitPointIdx + ((mod > splitPointIdx) ? splitPointIdx : mod) - 1;
}

後續操作也使用同樣的genSampleIndex函數來做選擇，這樣保證在操作所有序列上可以選取同樣的採樣點。

allValues = {ArrayList@10264}  size = 8  //本task有多少單詞
 0 = {Double@10266} -2.0
 1 = {Double@10271} -1.0
 2 = {Double@10272} -1.0
 3 = {Double@10273} -1.0
 4 = {Double@10274} -1.0
 5 = {Double@10275} -1.0
 6 = {Double@10276} -1.0
 7 = {Double@10277} -1.0
 
splitPoints = {ArrayList@10265}  size = 7 //採樣了7個
 0 = {Double@10266} -2.0
 1 = {Double@10271} -1.0
 2 = {Double@10272} -1.0
 3 = {Double@10273} -1.0
 4 = {Double@10274} -1.0
 5 = {Double@10275} -1.0
 6 = {Double@10276} -1.0

最後返回採樣數據，返回時候附帶當前taskIDnew Tuple2 <Object, Integer>(obj,taskId)。

這裡有一個trick點

  for (Object obj : splitPoints) {
     Tuple2 <Object, Integer> cur
        = new Tuple2 <Object, Integer>(
        obj,
        taskId); //這裏返回的是類似 (-5.0,2) ：其中2就是task id，-5.0是-wordcount。
     out.collect(cur);
  }

  out.collect(new Tuple2(
     getRuntimeContext().getNumberOfParallelSubtasks(),
     -taskId - 1));//這裏返回的是一個特殊元素，類似(4,-2) ：其中4是本應用中并行task數目，-2是當前-taskId - 1。這個task數目後續就會用到。

具體數據參見如下：

row = {Row@10211} "中國,-1.0,1.0986122886681098"
 fields = {Object[3]@10214} 
 
cur = {Tuple2@10286} "(-5.0,2)" // 返回採樣數據，返回時候附帶當前taskID
 f0 = {Double@10285} -5.0 // -wordcount。
 f1 = {Integer@10300} 2 // 當前taskID

歸併 SplitPointReducer

歸併所有task生成的sample。然後再次sample，把sample數據組成一個數據塊，這個數據塊選擇的原則是：每個task都盡量選擇若干sample。

這裏其實是有一個轉換，就是從正常單詞的抽樣 轉換到 某一類單詞的抽樣，這某一類的意思舉例是：出現次數為一，或者出現次數為五 這種單詞。

這裏all是所有採樣數據，其中一個元素內容舉例 (-5.0,2) ：其中2就是task id，-5.0是-wordcount。

這裏用 Collections.sort(all, new PairComparator()); 來對所有採樣數據做排序。排序基準是首先對 -wordcount，然後對task ID。

SplitPointReducer的返回採樣數值就作為廣播變量存儲起來：.withBroadcastSet(splitPoints, "splitPoints");

這裏的trick點是：

for (Tuple2 <Object, Integer> value : values) {
   if (value.f1 < 0) { 
      instanceCount = (int) value.f0;  // 特殊數據，類似(4,-2) ：其中4是本應用中task數目，這個就是後續選擇哪些taskid的基準
      continue;
   }
   all.add(new Tuple2 <>(value.f0, value.f1)); // (-5.0,2) 正常數據
}

選擇sample index splitPoints.add(allValues.get(index));也使用了同樣的genSampleIndex。

計算中具體數據如下：

for (int i = 0; i < splitPointSize; ++i) {
		int index = genSampleIndex(
					Long.valueOf(i),
					Long.valueOf(count),
					Long.valueOf(splitPointSize))
					.intValue();
		spliters.add(all.get(index));
}
for (Tuple2 <Object, Integer> spliter : spliters) {
		out.collect(spliter);
}

count = 33
all = {ArrayList@10245}  size = 33 // 所有採樣數據，
0 = {Tuple2@10256} "(-5.0,2)"// 2就是task id，-5.0是-wordcount。
1 = {Tuple2@10285} "(-2.0,0)"
......
6 = {Tuple2@10239} "(-1.0,0)"
7 = {Tuple2@10240} "(-1.0,0)"
8 = {Tuple2@10241} "(-1.0,0)"
9 = {Tuple2@10242} "(-1.0,0)"
10 = {Tuple2@10243} "(-1.0,0)"
11 = {Tuple2@10244} "(-1.0,1)"
......
16 = {Tuple2@10278} "(-1.0,1)"
......
24 = {Tuple2@10279} "(-1.0,2)"
......
32 = {Tuple2@10313} "(-1.0,3)"
  
// spliters是返回結果，這裏分別選取了all中index為8,16,24這個三個record。每個task都選擇了一個元素。
spliters = {HashSet@10246}  size = 3
 0 = {Tuple2@10249} "(-1.0,0)" // task 0 被選擇。就是說，這裏從task 0中選擇了一個count是1的元素，具體選擇哪個單詞其實不重要，就是為了選擇count是1的這種即可。
 1 = {Tuple2@10250} "(-1.0,1)" // task 1 被選擇。具體同上。
 2 = {Tuple2@10251} "(-1.0,2)" // task 2 被選擇。具體同上。

SplitData把真實數據IDF插入

use binary search to partition data into sorted subsets。前面函數給出的是詞的count，但是沒有IDF。這裏將用二分法查找 找到IDF，然後把IDF插入到partition data中。

首先要注意一點：splitData的輸入就是原始輸入input, 和splitPoints的輸入是一樣 的。

DataSet <Tuple2 <Integer, Row>> splitData = input
   .mapPartition(new SplitData(index))
   .withBroadcastSet(splitPoints, "splitPoints");

open函數中會取出廣播變量 splitPoints。

splitPoints = {ArrayList@10248}  size = 3
 0 = {Tuple2@10257} "(-1.0,0)"
 1 = {Tuple2@10258} "(-1.0,1)"
 2 = {Tuple2@10259} "(-1.0,2)"

本函數的輸入舉例

row = {Row@10232} "入門,-1.0,1.0986122886681098"

會在splitPoints中二分法查找，得到splits中每一個 sample 對應的真實IDF。然後發送出去。

這裏需要特殊說明下，這個二分法查找查找的是IDF數值，比如count為1的這種單詞對應的IDF數值，可能很多單詞都是count為1，所以找到一個這樣單詞的IDF即可。

splitPoints = {ArrayList@10223}  size = 3
 0 = {Tuple2@10229} "(-1.0,0)"
 1 = {Tuple2@10230} "(-1.0,1)"
 2 = {Tuple2@10231} "(-1.0,2)"
curTuple.f0 = {Double@10224} -1.0
  
int bsIndex = Collections.binarySearch(splitPoints, curTuple, new PairComparator());

		int curIndex;
		if (bsIndex >= 0) {
			curIndex = bsIndex;
		} else {
			curIndex = -bsIndex - 1;
		}

// 假設單詞是 "入門"，則發送的是 "入門" 這類單詞在本partition的index，和 "入門" 的單詞本身
// 其實，從調試過程看，是否發送單詞信息本身並不重要，因為接下來的那一步操作中，並沒有用到單詞本身信息
out.collect(new Tuple2 <>(curIndex, row)); 

reduceGroup計算同類型單詞數目

這裡是計算在某一partition中，某一種類單詞的數目。比如count為1的單詞，這種單詞總共有多少個。

後續會把new Tuple2 <>(id, count)作為partitionCnt廣播變量存起來。

id就是這類單詞在這partition中間的index，我們暫時稱之為partition index。count就是這類單詞在本partition的數目。

// 輸入舉例
value = {Tuple2@10312} "(0,入門,-1.0,1.0986122886681098)"
 f0 = {Integer@10313} 0
 
// 計算數目
for (Tuple2 <Integer, Row> value : values) {
		id = value.f0;
		count++;
}

out.collect(new Tuple2 <>(id, count));  
  
// 輸出舉例，假如是序號為0的這類單詞，其總體數目是12。這個序號0就是這類單詞在某一partition中的序號。就是上面的 curIndex。
id = {Integer@10313} 0
count = {Long@10338} 12

4.2.2 localSort

第二步是localSort。Sort a partitioned dataset. 最終排序並且會返回最終數值，比如 (29, “主編,-1.0,1.0986122886681098″)， 29就是”主編” 這個單詞在 IDF字典中的序號。

DataSet<Tuple2<Long, Row>> ordered = localSort(partitioned.f0, partitioned.f1, 1);

open函數中會獲取partitionCnt。然後計算出某一種類單詞，其在本partition之前所有partition中，這類單詞數目。

public void open(Configuration parameters) throws Exception {
		List <Tuple2 <Integer, Long>> bc = getRuntimeContext().getBroadcastVariable("partitionCnt");
		startIdx = 0L;
		int taskId = getRuntimeContext().getIndexOfThisSubtask();
		for (Tuple2 <Integer, Long> pcnt : bc) {
			if (pcnt.f0 < taskId) {
					startIdx += pcnt.f1;
			}
		}
}

bc = {ArrayList@10303}  size = 4
 0 = {Tuple2@10309} "(0,12)"  // 就是task0裏面，這種單詞有12個
 1 = {Tuple2@10310} "(2,9)"// 就是task1裏面，這種單詞有2個
 2 = {Tuple2@10311} "(1,7)"// 就是task2裏面，這種單詞有1個
 3 = {Tuple2@10312} "(3,9)"// 就是task3裏面，這種單詞有3個
// 如果本task id是4,則其startIdx為30。就是所有partition之中，它前面index所有單詞的和。  

然後進行排序。Collections.sort(valuesList, new RowComparator(field));

valuesList = {ArrayList@10405}  size = 9
 0 = {Row@10421} ":,-1.0,1.0986122886681098"
 1 = {Row@10422} "主編,-1.0,1.0986122886681098"
 2 = {Row@10423} "國內,-1.0,1.0986122886681098"
 3 = {Row@10424} "文獻,-1.0,1.0986122886681098"
 4 = {Row@10425} "李宜燮,-1.0,1.0986122886681098"
 5 = {Row@10426} "糖尿病,-1.0,1.0986122886681098"
 6 = {Row@10427} "美國,-1.0,1.0986122886681098"
 7 = {Row@10428} "謝恩,-1.0,1.0986122886681098"
 8 = {Row@10429} "象棋,-1.0,1.0986122886681098"
  
  
// 最後返回時候，就是  (29, "主編,-1.0,1.0986122886681098")，29就是“主編”這個單詞在最終字典中的序號。
// 這個序號是startIdx + cnt，startIdx是某一種類單詞，其在本partition之前所有partition中，這類單詞數目。比如在本partition之前，這類單詞有28個，則本partition中，從29開始計數。就是最終序列號
	for (Row row : valuesList) {
		out.collect(Tuple2.of(startIdx + cnt, row));
		cnt++; // 這裏就是在某一類單詞中，單調遞增，然後賦值一個字典序列而已
	}  
cnt = 1
row = {Row@10336} "主編,-1.0,1.0986122886681098"
 fields = {Object[3]@10339} 
startIdx = 28

4.3 過濾

最後還要進行過濾，如果文字個數超出了字典大小，就拋棄多餘文字。

ordered.filter(new FilterFunction<Tuple2<Long, Row>>() {
    @Override
    public boolean filter(Tuple2<Long, Row> value) {
        return value.f0 < vocabSize;
    }
})

0x05 生成模型

具體生成模型代碼如下。

DataSet<DocCountVectorizerModelData> resDocCountModel = ordered.filter(new FilterFunction<Tuple2<Long, Row>>() {
    @Override
    public boolean filter(Tuple2<Long, Row> value) {
        return value.f0 < vocabSize;
    }
}).mapPartition(new BuildDocCountModel(params)).setParallelism(1);
return resDocCountModel;

其中關鍵類是 DocCountVectorizerModelData 和 BuildDocCountModel。

5.1 DocCountVectorizerModelData

這是向量信息。

/**
 * Save the data for DocHashIDFVectorizer.
 *
 * Save a HashMap: index(MurMurHash3 value of the word), value(Inverse document frequency of the word).
 */
public class DocCountVectorizerModelData {
    public List<String> list;
    public String featureType;
    public double minTF;
}

5.2 BuildDocCountModel

最終生成的模型信息如下，這個也就是之前樣例代碼給出的輸出。

modelData = {DocCountVectorizerModelData@10411} 
 list = {ArrayList@10409}  size = 37
  0 = "{"f0":"9787310003969","f1":1.0986122886681098,"f2":19}"
  1 = "{"f0":"下冊","f1":1.0986122886681098,"f2":20}"
  2 = "{"f0":"全","f1":1.0986122886681098,"f2":21}"
  3 = "{"f0":"華齡","f1":1.0986122886681098,"f2":22}"
  4 = "{"f0":"圖解","f1":1.0986122886681098,"f2":23}"
  5 = "{"f0":"思","f1":1.0986122886681098,"f2":24}"
  6 = "{"f0":"成像","f1":1.0986122886681098,"f2":25}"
  7 = "{"f0":"舊書","f1":1.0986122886681098,"f2":26}"
  8 = "{"f0":"索引","f1":1.0986122886681098,"f2":27}"
  9 = "{"f0":":","f1":1.0986122886681098,"f2":28}"
  10 = "{"f0":"主編","f1":1.0986122886681098,"f2":29}"
  11 = "{"f0":"國內","f1":1.0986122886681098,"f2":30}"
  12 = "{"f0":"文獻","f1":1.0986122886681098,"f2":31}"
  13 = "{"f0":"李宜燮","f1":1.0986122886681098,"f2":32}"
  14 = "{"f0":"糖尿病","f1":1.0986122886681098,"f2":33}"
  15 = "{"f0":"美國","f1":1.0986122886681098,"f2":34}"
  16 = "{"f0":"謝恩","f1":1.0986122886681098,"f2":35}"
  17 = "{"f0":"象棋","f1":1.0986122886681098,"f2":36}"
  18 = "{"f0":"二手","f1":0.0,"f2":0}"
  19 = "{"f0":"）","f1":0.6931471805599453,"f2":1}"
  20 = "{"f0":"/","f1":1.0986122886681098,"f2":2}"
  21 = "{"f0":"出版社","f1":0.6931471805599453,"f2":3}"
  22 = "{"f0":"（","f1":0.6931471805599453,"f2":4}"
  23 = "{"f0":"入門","f1":1.0986122886681098,"f2":5}"
  24 = "{"f0":"醫學","f1":1.0986122886681098,"f2":6}"
  25 = "{"f0":"文集","f1":1.0986122886681098,"f2":7}"
  26 = "{"f0":"正版","f1":1.0986122886681098,"f2":8}"
  27 = "{"f0":"版","f1":1.0986122886681098,"f2":9}"
  28 = "{"f0":"電磁","f1":1.0986122886681098,"f2":10}"
  29 = "{"f0":"選讀","f1":1.0986122886681098,"f2":11}"
  30 = "{"f0":"中國","f1":1.0986122886681098,"f2":12}"
  31 = "{"f0":"書","f1":1.0986122886681098,"f2":13}"
  32 = "{"f0":"十二冊","f1":1.0986122886681098,"f2":14}"
  33 = "{"f0":"南開大學","f1":1.0986122886681098,"f2":15}"
  34 = "{"f0":"文學","f1":1.0986122886681098,"f2":16}"
  35 = "{"f0":"郁達夫","f1":1.0986122886681098,"f2":17}"
  36 = "{"f0":"館藏","f1":1.0986122886681098,"f2":18}"
 featureType = "WORD_COUNT"
 minTF = 1.0

0x06 預測

預測業務邏輯是DocCountVectorizerModelMapper

首先我們可以看到 FeatureType，這個可以用來配置輸出哪種信息。比如可以輸出以下若干種：

public enum FeatureType implements Serializable {
    /**
     * IDF type, the output value is inverse document frequency.
     */
    IDF(
        (idf, termFrequency, tokenRatio) -> idf
    ),
    /**
     * WORD_COUNT type, the output value is the word count.
     */
    WORD_COUNT(
        (idf, termFrequency, tokenRatio) -> termFrequency
    ),
    /**
     * TF_IDF type, the output value is term frequency * inverse document frequency.
     */
    TF_IDF(
        (idf, termFrequency, tokenRatio) -> idf * termFrequency * tokenRatio
    ),
    /**
     * BINARY type, the output value is 1.0.
     */
    BINARY(
        (idf, termFrequency, tokenRatio) -> 1.0
    ),
    /**
     * TF type, the output value is term frequency.
     */
    TF(
        (idf, termFrequency, tokenRatio) -> termFrequency * tokenRatio
    );
}

其次，在open函數中，會加載模型，比如：

wordIdWeight = {HashMap@10838}  size = 37
 "醫學" -> {Tuple2@10954} "(6,1.0986122886681098)"
 "選讀" -> {Tuple2@10956} "(11,1.0986122886681098)"
 "十二冊" -> {Tuple2@10958} "(14,1.0986122886681098)"
...
 "華齡" -> {Tuple2@11022} "(22,1.0986122886681098)"
 "索引" -> {Tuple2@11024} "(27,1.0986122886681098)"
featureType = {DocCountVectorizerModelMapper$FeatureType@10834} "WORD_COUNT"

最後，預測時候調用predictSparseVector函數，會針對輸入 二手 舊書 : 醫學 電磁 成像來進行匹配。生成稀疏向量SparseVector。

0|$37$0:1.0 6:1.0 10:1.0 25:1.0 26:1.0 28:1.0

以上表示那幾個單詞 分別對應0 6 10 25 26 28 這幾個字典中對應序號的單詞，其在本句對應的出現數目都是一個。

0x07 參考

Tf-Idf詳解及應用

https://github.com/fxsjy/jieba

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理

【其他文章推薦】

※產品缺大量曝光嗎?你需要的是一流包裝設計!

※自行創業缺乏曝光? 網頁設計幫您第一時間規劃公司的形象門面

※回頭車貨運收費標準

※推薦評價好的iphone維修中心

※超省錢租車方案

※台中搬家遵守搬運三大原則,讓您的家具不再被破壞!

※推薦台中搬家公司優質服務,可到府估價