大規模異常濫用檢測：基於局部敏感哈希算法——來自Uber Engineering的實踐

 

uber全球用戶天天會產生500萬條行程，保證數據的準確性相當重要。若是全部的數據都獲得有效利用，t經過元數據和聚合的數據能夠快速檢測平臺上的濫用行爲，如垃圾郵件、虛假帳戶和付款欺詐等。放大正確的數據信號能使檢測更精確，也所以更可靠。

爲了解決咱們和其餘系統中的相似挑戰，Uber Engineering 和 Databricks 共同向Apache Spark 2.1開發了局部敏感哈希（LSH）。LSH是大規模機器學習中經常使用的隨機算法和哈希技術，包括聚類和近似最近鄰搜索。

在這篇文章中，咱們將講解Uber如何使用這個強大的工具進行大規模的欺詐行程檢測。

爲何使用LSH？

在 Uber Engineering 實現 LSH 以前，咱們篩選行程的算法複雜度爲 N^2； 儘管精度較高，N^2 的算法複雜度對於 Uber 當前的數據規模過於耗時、密集（volume-intensive），且對硬件要求高。

LSH的整體思路是使用一系列函數（稱爲 LSH 族）將數據點哈希到桶(buckets)中，使距離較近的數據點位於同一個桶中的機率較高，而距離很遠的數據點在不一樣的桶裏。所以, LSH 算法能使具備不一樣程度重疊行程的識別更爲容易。

做爲參考，LSH 是一項有大量應用方向的多功能技術，其中包括：

• 近似重複的檢測： LSH 一般用於對大量文檔，網頁和其餘文件進行去重處理。
• 全基因組的相關研究：生物學家常用 LSH 在基因組數據庫中鑑定類似的基因表達。
• 大規模的圖片搜索： Google 使用 LSH 和 PageRank 來構建他們的圖片搜索技術VisualRank。
• 音頻/視頻指紋識別：在多媒體技術中，LSH 被普遍用於 A/V 數據的指紋識別。

LSH 在 Uber 的應用

LSH 在 Uber 主要用於欺詐司機的判斷，基於空間特性檢測類似的行程。Uber 工程師在2016年Spark峯會上介紹了這個用例，討論咱們團隊在Spark框架中使用LSH的動機，以便結合全部行程數據並從中篩選欺詐行爲。咱們在Spark上使用LSH的動機有三個方面：

1. Spark是Uber運營的重要組成部分，許多內部團隊目前使用Spark進行機器學習、空間數據處理、時間序列計算、分析與預測以及特別的數據科學探索等各類複雜的數據處理。實際上，Uber 在YARN和Mesos上都使用了幾乎全部的Spark組件，如MLlib，Spark SQL，Spark Streaming和直接RDD處理; 因爲咱們的基礎架構和工具圍繞Spark構建，咱們的工程師能夠輕鬆建立和管理Spark應用程序。
2. 在具體的機器學習進行以前，Spark能夠高效地進行數據清理和特徵工程(feature engineering)，使數據處理速度極大的提升。Uber收集的大量數據使傳統方法解決此問題時難以擴展且速度較慢。
3. 咱們不須要方程的精確解，所以不須要購買和維護額外的硬件。近似值爲咱們提供了足夠的信息來判斷是否存在潛在的欺詐活動，在這種狀況下，這些信息足以解決咱們的問題。LSH容許咱們犧牲一些精度來節省大量的硬件資源。

出於這些緣由，在Spark上部署LSH解決此問題是達到咱們業務目標的正確選擇：可擴展，數據規模和精度。(譯註：原文爲scale, scale and scale again)

在更高的層面上，咱們使用LSH方法有三個步驟。首先，咱們經過將每一個行程分解爲相同大小的區域段,爲其建立一個特徵向量。而後，咱們對Jaccard距離函數使用用MinHash哈希這些特徵向量。最後，咱們實時的使用批量類似度鏈接(similarity join in batch)或k-Nearest Neighbor搜索。與檢測欺詐的簡單暴力算法相比，咱們當前的數據集下Spark工做的完成速度提升了整個數量級（從使用N^2方法的約55小時到使用LSH約4小時）。

API教程

爲了最好地展現LSH的工做原理，咱們將在Wikipedia Extraction（WEX）數據集上使用MinHashLSH 尋找類似的文章。

每一個LSH家族都與其度量空間相關。在Spark 2.1中，有兩個LSH估計器：

• 基於歐幾里德距離的BucketedRandomProjectionLSH
• 基於Jaccard距離的MinHashLSH

咱們須要對詞數的實特徵向量進行處理，所以,這種狀況下咱們選擇使用MinHashLSH。

加載原始數據

首先，咱們須要啓動一個EMR（Elastic MapReduce彈性MapReduce）集羣，並將WEX數據集掛載爲一個EBS（Elastic Block Store 彈性塊存儲）卷。此過程額外的細節能夠經過亞馬遜的EMR和EBS相關文檔。

在創建Spark集羣並掛載WEX數據集後，咱們根據集羣大小將一個WEX數據樣本上傳到HDFS。在Spark shell中，咱們加載HDFS樣本數據:

// Read RDD from HDFS
import org.apache.spark.ml.feature._ import org.apache.spark.ml.linalg._ import org.apache.spark.sql.types._ val df = spark.read.option("delimiter","\t").csv("/user/hadoop/testdata.tsv") val dfUsed = df.select(col("_c1").as("title"), col("_c4").as("content")).filter(col("content") !== null) dfUsed.show()

圖1：維基百科中的文章以標題和內容表示。

圖1顯示了咱們上方代碼的結果，按標題和內容顯示文章。咱們將使用該內容做爲咱們的哈希鍵，並在後面的實驗中大體找到相似的維基百科文章。

準備特徵向量

MinHash用於快速估計兩個數據集的類似度，是一種很是常見的LSH技術。在Spark中實現的MinHashLSH，咱們將每一個數據集表示爲一個二進制稀疏向量。在這一步中，咱們將把維基百科文章的內容轉換成向量。

使用如下代碼進行特徵工程，咱們將文章內容分割成單詞（Tokenizer），建立單詞計數的特徵向量（CountVectorizer），並刪除空的文章：

// Tokenize the wiki content
val tokenizer = new Tokenizer().setInputCol("content").setOutputCol("words") val wordsDf = tokenizer.transform(dfUsed) // Word count to vector for each wiki content val vocabSize = 1000000 val cvModel: CountVectorizerModel = new CountVectorizer().setInputCol("words").setOutputCol("features").setVocabSize(vocabSize).setMinDF(10).fit(wordsDf) val isNoneZeroVector = udf({v: Vector => v.numNonzeros > 0}, DataTypes.BooleanType) val vectorizedDf = cvModel.transform(wordsDf).filter(isNoneZeroVector(col("features"))).select(col("title"), col("features")) vectorizedDf.show()

圖2：在對代碼進行特徵工程以後，維基百科文章的內容被轉換爲二進制稀疏矢量。

LSH模型的擬合和查詢

爲了使用MinHashLSH，咱們首先用下面的命令，在咱們的特徵數據上擬合一個MinHashLSH模型：

val mh = new MinHashLSH().setNumHashTables(3).setInputCol("features").setOutputCol("hashValues") val model = mh.fit(vectorizedDf)

咱們能夠用咱們的LSH模型進行多種類型的查詢，但爲了本教程的目的，咱們首先對數據集執行一次特徵轉換：

model.transform(vectorizedDf).show()

這個命令爲咱們提供了哈希值，有利於手動鏈接(manual joins)和特徵生成。

圖3： MinHashLSH添加了一個新列來存儲哈希。每一個哈希表示爲一個向量數組。

接下來，咱們執行一個近似最近鄰（Approximate Nearest Neighbor,ANN）搜索，以找到離咱們目標最近的數據點。出於演示的目的，咱們搜索的內容可以大體匹配"united states"的文章。

val key = Vectors.sparse(vocabSize, Seq((cvModel.vocabulary.indexOf("united"), 1.0), (cvModel.vocabulary.indexOf("states"), 1.0))) val k = 40 model.approxNearestNeighbors(vectorizedDf, key, k).show()

圖4：近似最近鄰搜索結果，查找維基百科中有關「united states」的文章。

最後，咱們運行一個近似類似鏈接（approximate similarity join），在同一個數據集中找到類似的文章對：

// Self Join
val threshold = 0.8 model.approxSimilarityJoin(vectorizedDf, vectorizedDf, threshold).filter("distCol != 0").show()

雖然咱們在下面使用自鏈接，但咱們也能夠鏈接不一樣的數據集來獲得相同的結果。

圖5：近似類似鏈接列出了相似的維基百科文章，並設置哈希表的數量。

圖5演示瞭如何設置哈希表的數量。對於一個近似類似鏈接和近似最近鄰命令，哈希表的數量能夠平衡運行時間和錯誤率（OR-amplification）。增長哈希表的數量會提升準確性，但也會增長程序的通訊成本和運行時間。默認狀況下，哈希表的數量設置爲1。

想要在Spark 2.1中進行其它使用LSH的練習，還能夠在Spark發佈版中運行和BucketRandomProjectionLSH、MinHashLSH相關的更小示例。

性能測試

爲了衡量性能，咱們在WEX數據集上測試了MinHashLSH的實現。使用AWS雲，咱們使用16個executors（m3.xlarge 實例）執行WEX數據集樣本的近似最近鄰搜索和近似類似鏈接。

圖6：使用numHashTables = 5，近似最近鄰的速度比徹底掃描快2倍。在numHashTables = 3的狀況下，近似類似鏈接比徹底鏈接和過濾要快3-5倍。

在上面的表格中，咱們能夠看到哈希表的數量被設置爲5時，近似最近鄰的運行速度徹底掃描快2倍;根據不一樣的輸出行和哈希表數量，近似類似鏈接的運行速度快了3到5倍。

咱們的實驗結果還代表，儘管當前算法的運行時間很短，但與暴力方法的結果相比仍有較高的精度。近似最近鄰搜索對於40個返回行達到了85％的正確率，而咱們的近似類似鏈接成功地找到了93％的鄰近行。這種速度與精度的折中算法，證實了LSH能從天天TB級數據中檢測欺詐行爲的強大能力。

下一步

儘管咱們的LSH模型可以幫助Uber識別司機的欺詐行爲，但咱們的工做還遠遠沒有完成。經過對LSH的初步實現，咱們計劃在將來的版本中添加一些新的功能。其中高優先級功能包括：

SPARK-18450：除了指定完成搜索所需的哈希表數量以外，這個新功能使用戶可以在每一個哈希表中定義哈希函數的數量。這個改變也將一樣提供對 AND/OR-compound增強的支持。

SPARK-18082＆SPARK-18083：咱們想要實現其餘的LSH familes函數。這兩個更新的實現將能對兩個數據點之間的漢明距離(Hamming distance)進行位採樣，並提供機器學習任務中經常使用的餘弦距離隨機投影符號。

SPARK-18454：第三個功能將改進近似最近鄰搜索的API。這種新的多探測（multi-probe ）類似性搜索算法，可以在不須要大量的哈希表的狀況下提高搜索的質量。

咱們將繼續開發和擴展當前項目，加入上述以及其餘的相關功能，很是歡迎你們的反饋。