基于HBase建構千億級文本資料相似度計算與快速去重系統

前言

随着大資料時代的到來，資料資訊在給我們生活帶來便利的同時，同樣也給我們帶來了一系列的考驗與挑戰。本文主要介紹了基于 Apache HBase 與 Google SimHash 等多種算法共同實作的一套支援百億級文本資料相似度計算與快速去重系統的設計與實作。該方案在公司業務層面徹底解決了多主題海量文本資料所面臨的存儲與計算慢的問題。

一. 面臨的問題

1. 如何選擇文本的相似度計算或去重算法？

常見的有餘弦夾角算法、歐式距離、Jaccard 相似度、最長公共子串、編輯距離等。這些算法對于待比較的文本資料不多時還比較好用，但在海量資料背景下，如果每天産生的資料以千萬計算，我們如何對于這些海量千萬級的資料進行高效的合并去重和相似度計算呢？

1. 如何實作快速計算文本相似度或去重呢？

如果我們選好了相似度計算和去重的相關算法，那我們怎麼去做呢？如果待比較的文本資料少，我們簡單周遊所有文本進行比較即可，那對于巨大的資料集我們該怎麼辦呢？周遊很明顯是不可取的。

1. 海量資料的存儲與快速讀寫

二. SimHash 算法引入

基于問題一，我們引入了 SimHash 算法來實作海量文本的相似度計算與快速去重。下面我們簡單了解下該算法。

1. 局部敏感哈希

在介紹 SimHash 算法之前，我們先簡單介紹下局部敏感哈希是什麼。局部敏感哈希的基本思想類似于一種空間域轉換思想，LSH 算法基于一個假設，如果兩個文本在原有的資料空間是相似的，那麼分别經過哈希函數轉換以後的它們也具有很高的相似度；相反，如果它們本身是不相似的，那麼經過轉換後它們應仍不具有相似性。

局部敏感哈希的最大特點就在于保持資料的相似性，舉一個小小的例子說明一下：對A文章微調後我們稱其為B文章（可能隻是多了一個‘的’字），如果此時我們計算兩篇文章的 MD5 值，那麼必将大相徑庭。而局部敏感哈希的好處是經過哈希函數轉換後的值也隻是發生了微小的變化，即如果兩篇文章相似度很高，那麼在算法轉換後其相似度也會很高。

MinHash 與 SimHash 算法都屬于局部敏感哈希，一般情況若每個 Feature 無權重，則 MinHash 效果優于 SimHash 有權重時 SimHash 合适。長文本使用 Simhash 效果很好，短文本使用 Simhash 準備度不高。

1. SimHash 算法

SimHash 是 Google 在2007年發表的論文《Detecting Near-Duplicates for Web Crawling 》中提到的一種指紋生成算法或者叫指紋提取算法，被 Google 廣泛應用在億級的重複網頁刪除偵測的 Job 中，其主要思想是降維，經過simhash降維後，可能僅僅得到一個長度為32或64位的二進制由01組成的字元串。而一維查詢則是非常快速的。

SimHash的工作原理我們這裡略過，大家可以簡單了解為：我們可以利用SimHash算法為每一個網頁/文章生成一個長度為32或64位的二進制由01組成的字元串（向量指紋），形如：1000010010101101111111100000101011010001001111100001001011001011。

1. 海明距離

兩個碼字的對應比特取值不同的比特數稱為這兩個碼字的海明距離。在一個有效編碼集中,任意兩個碼字的海明距離的最小值稱為該編碼集的海明距離。舉例如下：10101和00110從第一位開始依次有第一位、第四、第五位不同，則海明距離為3。

在 google 的論文給出的資料中，64位的簽名，在海明距離為3的情況下，可認為兩篇文檔是相似的或者是重複的，當然這個值隻是參考值。

這樣，基于 SimHash 算法，我們就可以将百億千億級的高維特征文章轉變為一維字元串後再通過計算其海明距離判斷網頁/文章的相似度，可想效率必将大大提高。

三. 效率問題

到這裡相似度問題基本解決，但是按這個思路，在海量資料幾百億的數量下，效率問題還是沒有解決的，因為資料是不斷添加進來的，不可能每來一條資料，都要和全庫的資料做一次比較，按照這種思路，處理速度會越來越慢，線性增長。

這裡，我們要引入一個新的概念：抽屜原理，也稱鴿巢原理。下面我們簡單舉例說一下：

桌子上有四個蘋果，但隻有三個抽屜，如果要将四個蘋果放入三個抽屜裡，那麼必然有一個抽屜中放入了兩個蘋果。如果每個抽屜代表一個集合，每一個蘋果就可以代表一個元素，假如有n+1個元素放到n個集合中去，其中必定有一個集合裡至少有兩個元素。

抽屜原理就是這麼簡單，那如果用它來解決我們海量資料的周遊問題呢？

針對海量資料的去重效率，我們可以将64位指紋，切分為4份16位的資料塊，根據抽屜原理在海明距離為3的情況，如果兩個文檔相似，那麼它必有一個塊的資料是相等的。

那也就是說，我們可以以某文本的 SimHash 的每個16位截斷指紋為 Key，Value 為 Key 相等時文本的 SimHash 集合存入 K-V 資料庫即可，查詢時候，精确比對這個指紋的4個16位截斷指紋所對應的4個 SimHash 集合即可。

如此，假設樣本庫，有2^37 條資料（1375億資料），假設資料均勻分布，則每個16位（16個01數字随機組成的組合為2^16 個）倒排傳回的最大數量為 (2^37) 4 / (2^16) =8388608個候選結果，4個16位截斷索引，總的結果為：48388608=33554432，約為3356萬，通過 這樣一來的降維處理，原來需要比較1375億次，現在隻需要比較3356萬次即可得到結果，這樣以來大大提升了計算效率。

根據網上測試資料顯示，普通 PC 比較1000萬次海明距離大約需要 300ms，也就是說3356萬次（1375億資料）隻需花費3356/1000*0.3=1.0068s。那也就是說對于千億級文本資料（如果每個文本1kb，約100TB資料）的相似度計算與去重工作我們最多隻需要一秒的時間即可得出結果。

四. HBase 存儲設計

饒了這麼大一周，我們終于将需要講明的理論知識給大家過了一遍。為了闡述的盡量清晰易懂，文中很多理論知識的了解借鑒了大量部落客大牛的部落格，原文連結已在文末附上，有不太明白的地方快快跪拜大牛們的部落格吧，哈哈！

下面我們着重介紹一下 HBase 存儲表的設計與實作。

基于上文我們可以大概知道，如果将64位指紋平分四份，海明距離取3，那麼必有一段16位截取指紋的資料是相等的。而每一段16位截取指紋對應一個64位指紋集合，且該集合中的每個64位指紋必有一段16位截取指紋與該段16位截取指紋重合。我們可以簡單表示(以8位非01指紋舉例)為：

key value(set) 12 [12345678,12345679] 23 [12345678,12345679,23456789]

那如果基于 HBase 去實作的話，我們大概對比三種可能的設計方案。

方案一：

以 16 位指紋作為 HBase 資料表的行鍵，将每一個與之可能相似的64位指紋作為 HBase 的列，列值存文章id值，即建構一張大寬表。如下表所示(以8位非01指紋舉例)：

rowkey column1 column2 column3 ...

實際資料表可能是這個樣子：

rowkey 12345678 32234567 23456789 12456789 ... 12 1102101 1102102 ... 23 1102104 1102105 ... 34 1102106 ...

那其實這樣設計表的話該 HBase 表 Rowkey 的個數就是一個确定的數值：16個01數字随機組成的組合為2^16 個。也就是共2^16=65536行。 列的個數其實也是固定的，即2^64=184467440737億萬列。

此時，比如說我們比較56431234與庫中所有文本的相似度，隻需拉去rowkey in (56,43,12,34) 四行資料周遊每行列，由于 HBase 空值不進行存儲，所有隻會周遊存在值的列名。

由上文我們計算出1350億資料如果平均分布的話每行大約有839萬列，且不說我們的資料量可能遠遠大于千億級别，也不說以64位字元串作為列名所占的存儲空間有多大，單單千億級資料量 HBase 每行就大約839萬列，雖說HBase号稱支援千萬行百萬列資料存儲，但總歸還是設計太不合理。資料不會理想化均勻分布，總列數高達184467440737億萬列也令人堪憂。

方案二：

以 16 位指紋與64位指紋拼接後作為 HBase 資料表的行鍵，該表隻有一列，

買手機遊戲平台

列值存文章id值，即建構一張大長表。如下表所示(以8位非01指紋舉例)：

rowkey id

rowkey id 12_12345678 1 34_12345678 1 56_12345678 1 78_12345678 1 34_22345678 2 23_12235678 3

如此設計感覺要比第一種方法要好一些，每一篇文章會被存為四行。但同樣有諸多缺點，一是 Rowkey 過長，二是即便我們通過某種轉變設計解決了問題一，那擷取資料時我們也隻能将 Get 請求轉為四個Scan并發掃描+StartEnKey 去掃描表擷取資料。當然，如果想實作順序掃描還可能存在熱點問題。在存儲上，也造成了資料大量備援。

方案三：

在真實生産環境中，我們采取該方案來避免上述兩個方案中出現的問題與不足。下面簡單介紹一下（如果您有更好更優的方案，歡迎留言，先表示感謝！）

簡言之呢，就是自己在 HBase 端維護了一個 Set 集合（協處理器），并以 Json 串進行存儲，格式如下：

{

"64SimHash1":"id1",
"64SimHash2":"id2",
...
...           

}

基于公司存在多種主題類型的文本資料，且互相隔離，去重與相似度計算也是分主題進行，我們的 Rowkey 設計大緻如下：

Rowkey = HashNumber_ContentType_16SimHash (共24位)

HashNumber： 為防熱點，對表進行Hash預分區（64個預分區），占2個字元 計算公式如下：String.format("%02x", Math.abs(key.hashCode()) % 64)

ContentType ：内容主題類型，占4個字元

16SimHash： 16位 SimHash 截取指紋，由01組成

表結構大緻如下：

rowkey si s0 s1 s2 s3 ... 01_news_010101010101010101 value 1 Json 串 ... 02_news_010101010101010110 value 2 Json 串 Json 串 ... 03_news_100101010101010110 value 3 Json 串 Json 串 Json 串 ... 01_xbbs_010101010101010101 value 1 Json 串 ...

si：用戶端傳遞過來的欲存儲的值，由64位 Simhash 與 Id 通過雙下劃線拼接而成，諸如 Simhash__Id 的形式。 s0：記錄該行資料共有多少個 Set 集合，每一個 Set 集合存儲10000個K-V對兒（約1MB）。 s1：第一個 Set 集合，Json 串存儲，如果 Size > 10000 ，之後來的資料将存入s2。 s2：以此類推。

當然最核心的部分是s1/s2/s3 中 Json 串中要排重。最簡單的辦法無非是每次存入資料前先将所有 Set 集合中的資料讀到用戶端，将欲存的資料與集合中所有資料比對後再次插入。這将帶來大量往返IO開銷，影響寫性能。是以，我們在此引入了 HBase 協處理器技術來規避這個問題，即在服務端完成所有排重操作。大緻代碼如下：

package com.learn.share.scenarios.observers;

import com.google.gson.Gson;

import com.google.gson.JsonObject;

import org.apache.commons.lang.StringUtils;

import org.apache.hadoop.hbase.Cell;

import org.apache.hadoop.hbase.CellUtil;

import org.apache.hadoop.hbase.CoprocessorEnvironment;

import org.apache.hadoop.hbase.client.Durability;

import org.apache.hadoop.hbase.client.Get;

import org.apache.hadoop.hbase.client.Put;

import org.apache.hadoop.hbase.client.Result;

import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;

import org.apache.hadoop.hbase.coprocessor.ObserverContext;

import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;

import org.apache.hadoop.hbase.regionserver.wal.WALEdit;

import org.apache.hadoop.hbase.util.Bytes;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import java.io.IOException;

import java.util.List;

/**

• 基于協處理器建構百億級文本去重系統

*/

public class HBaseSimHashSetBuildSystem extends BaseRegionObserver {

private Logger logger = LoggerFactory.getLogger(HBaseSimHashSetBuildSystem.class);


@Override
public void start(CoprocessorEnvironment e) throws IOException {
    logger.info("Coprocessor opration start...");
}

/**
 *
 * @param e
 * @param put
 * @param edit
 * @param durability
 * @throws IOException
 */
@Override
public void prePut(ObserverContext<RegionCoprocessorEnvironment> e, Put put, WALEdit edit, Durability durability) throws IOException {
    // test flag
    logger.info("do something before Put Opration...");

    List<Cell> cells = put.get(Bytes.toBytes("f"), Bytes.toBytes("si"));
    if (cells == null || cells.size() == 0) {
        return;
    }
    String simhash__itemid = Bytes.toString(CellUtil.cloneValue(cells.get(0)));
    if (StringUtils.isEmpty(simhash__itemid)||simhash__itemid.split("__").length!=2){
        return;
    }
    String simhash = simhash__itemid.trim().split("__")[0];
    String itemid = simhash__itemid.trim().split("__")[1];

    // 擷取Put Rowkey
    byte[] row = put.getRow();
    // 通過Rowkey構造Get對象
    Get get = new Get(row);
    get.setMaxVersions(1);
    get.addFamily(Bytes.toBytes("f"));
    Result result = e.getEnvironment().getRegion().get(get);
    Cell columnCell = result.getColumnLatestCell(Bytes.toBytes("f"), Bytes.toBytes("s0")); // set size
    if (columnCell == null) {
        // 第一次存儲資料，将size初始化為1
        logger.info("第一次存儲資料，将size初始化為1");

        JsonObject jsonObject = new JsonObject();
        jsonObject.addProperty(simhash,itemid);
        Gson gson = new Gson();
        String json = gson.toJson(jsonObject);

        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("s1"), Bytes.toBytes(json)); // json 數組
        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("s0"), Bytes.toBytes("1"));  // 初始化
    }else {
        byte[] sizebyte = CellUtil.cloneValue(columnCell);
        int size = Integer.parseInt(Bytes.toString(sizebyte));
        logger.info("非第一次存儲資料 ----> Rowkey `"+Bytes.toString(row)+"` simhash set size is : "+size +", the current value is : "+simhash__itemid);
        for (int i = 1; i <= size; i++) {
            Cell cell1 = result.getColumnLatestCell(Bytes.toBytes("f"), Bytes.toBytes("s"+i));
            String jsonBefore = Bytes.toString(CellUtil.cloneValue(cell1));
            Gson gson = new Gson();
            JsonObject jsonObject = gson.fromJson(jsonBefore, JsonObject.class);
            int sizeBefore = jsonObject.entrySet().size();
            if(i==size){
                if(!jsonObject.has(simhash)){
                    if (sizeBefore==10000){
                        JsonObject jsonone = new JsonObject();
                        jsonone.addProperty(simhash,itemid);
                        String jsonstrone = gson.toJson(jsonone);
                        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("s"+(size+1)), Bytes.toBytes(jsonstrone)); // json 數組
                        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("s0"), Bytes.toBytes((size+1)+""));  // 初始化
                    }else {
                        jsonObject.addProperty(simhash,itemid);
                        String jsonAfter = gson.toJson(jsonObject);
                        put.addColumn(Bytes.toBytes("f"),Bytes.toBytes("s"+size), Bytes.toBytes(jsonAfter)); // json 數組
                    }
                }else {
                    return;
                }
            }else{
                if(!jsonObject.has(simhash)){
                    continue;
                }else {
                    return;
                }
            }
        }
    }
}           

如此，當我們需要對某一文本指紋與庫中資料進行比對時，隻需一個Table.Get(List) 操作即可傳回所有的資料，然後基于s0依次擷取各個 Set 集合中的資料即可。

下面我們算一筆賬，假設我們某主題類型資料依然有 2^37 條資料（1375億資料），假設資料均勻分布，則每個16位（16個01數字随機組成的組合為2^16 個）倒排傳回的最大數量為 (2^37) * 4 / (2^16) =8388608個候選結果，即每行約839個 Set 集合，每個Set 集合大約1M 的話，資料存儲量也必然不會太大。

你如果有十種不同主題的資料，HBase 行數無非也才 (2^16)*10 = 655360 行而已。

如果再加上 Snappy 壓縮呢？ 如果再加上 Fast-Diff 編碼呢？ 如果再開啟 Mob 對象存儲呢？ 每個 Set 是不是可以存10萬個鍵值對？每行隻需90個 Set 集合。

也或許，如果資料量小的話，使用 Redis 是不是更好呢？

總之，優化完善和不完美的地方還很多，本文也就簡單叙述到此，如果您有好的建議或是不同看法，歡迎留言哦！感恩~ 晚安各位~~