大规模异常滥用检测：基于局部敏感哈希算法——来自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 ）类似性搜索算法，可以在不须要大量的哈希表的状况下提高搜索的质量。

咱们将继续开发和扩展当前项目，加入上述以及其余的相关功能，很是欢迎你们的反馈。