【IT168 资讯】了解如何使用Java和Spark MLlib开发一种算法，该算法能够根据700万条记录的数据集检测欺诈行为。

　　在这篇文章中，我们将使用Spark MLlib开发Java中的算法。完整的工作代码可以从GitHub下载。可以在不使用深入的Java知识（使用配置文件）的情况下，用几种不同的配置和实验来运行代码。

　　在之前的文章中，我们使用Octave实现了相同的异常检测算法。我们从七百万个筛选出了500,000条记录（仅限于TRANSFER类型），以便调查和了解可用数据。此外，还绘制了几张图表来显示数据和异常（欺诈）的样子。由于Octave加载了内存中的所有数据，因此它对大数据有限制。出于这个原因，我们将使用Spark在700万个更大的数据集上运行异常检测。

　　高斯分布

　　本节简要介绍如何使用高斯函数进行异常检测。高斯密度函数具有钟形曲线形状，如下所示：

　　大部分数据的常规数据往往处于钟形曲线的中心，而边缘的异常更为罕见。与此同时，我们可以看到边缘上的点与中心点（接近0.4）相比具有更低的函数值（或者概率小于0.1）。

　　在这个例子之后，我们可以说每一个具有低于0.05的概率密度函数的例子都是异常的。当然，我们可以根据需要来控制阈值。大的值意味着更多的异常被标记，其中大部分可能不是异常。另一方面，小的值意味着我们可能错过异常，因为算法变得更加宽容。

　　上面的例子是一维的，数据只有一个特征。实际上，我们有更多的功能和维度的数据。为了将数据绘制到图中，我们使用主成分分析（PCA）将数据的维数减少到二维（2D）甚至三维（3D）。以下是两个维度的示例：　　

　　注意正常的数据在第一个和第二个圆的中间趋于一起，异常在第三个圆的边缘。图上的圆圈表示高斯钟形曲线如何在数据之间分布（通常，它将在3D中为钟形，但为了简单明了，以2D表示）。

　　为了在钟形图中的某个位置上举一个例子，我们需要计算两个分量：μ（均值）和σ2（方差）。一旦计算了均值和方差，我们可以应用一个相当简单的公式来得到新的例子的密度概率。如果概率低于某个特定值（σ），我们将其标记为异常;否则，这是正常的。在我以前的文章中查找关于开发的细节。

　　Spark和MLlib

　　本节提供Spark和MLlib的简要说明。

　　Spark

　　Apache Spark是一个集群计算框架。 Spark帮助我们在群集中的不同节点上并行执行作业，然后将这些结果合并成一个结果/响应。它将我们的数据集合转换为分布在集群节点（称为RDD（弹性分布式数据集））的元素集合。例如，在一个Java程序中，我们可以将一个集合转换成一个能够并行操作的RDD，如下所示：

　　并行集合被分割成分区，Spark的每个分区执行一个任务，所以我们希望每个CPU有两到四个分区。我们可以通过用sc.parallelize（collection，partitionNumber）定义另一个参数来控制Spark创建的分区数量。除了来自应用程序的集合之外，Spark还能够转换来自Hadoop支持的存储源（包括本地文件系统，HDFS，Cassandra，HBase和Amazon S3）的数据。

　　将数据转换为RDD后，我们可以在集群节点上执行两种并行操作。转换操作将RDD集合作为输入，并返回一个新的RDD集合，如映射和操作，它们采用RDD并返回单个结果，如reduce、count等。不管类型如何，操作都是惰性的，类似于Java 8在定义时不运行，而是在请求时运行。因此，可以在请求时多次计算一个操作，为了避免这种情况，保存在内存或缓存中。

　　MLlib

　　Spark支持Java、Scala、Python和R API。它还支持一套丰富的高级工具，包括用于SQL和结构化数据处理的Spark SQL，用于机器学习的MLlib，用于图形处理的GraphX以及Spark Streaming。

　　MLlib是Spark的机器学习（ML）库。它提供了几个现成的ML工具，如：

　　ML算法

　　·分类

　　·回归

　　·聚类

　　·协作过滤

　　Featurization

　　·特征提取

　　·转型

　　·降维

　　·选择

　　公用事业

　　·线性代数

　　·统计

　　·数据处理

　　数据准备

　　我们需要为算法执行准备数据。以下是数据的样子：　　

　　我们需要把所有东西都转换成数字。 幸运的是，大部分数据都是数字，只有nameOrig和nameDest以C，D或M这样的字符开始。我们简单地用1代替C，用2代替D，用3代替M。同样，我们将字符从chars转换为数字 如下所示：

　　所有的准备工作都是通过使用Spark转换操作映射的Java代码完成的：　

　　之后，文件应该是这样的：

　　由于较大的文件大小和GitHub文件大小限制，数据不在代码中提供。你可以从这里下载文件（https://www.kaggle.com/ntnu-testimon/paysim1），将其重命名为allData.csv（将常量FILE_NAME更改为其他名称），并将其复制到文件夹data /中。

　　执行算法

　　让我们一步一步看看我们如何执行异常检测算法。

　　1.从所有数据（七百万）中，我们需要随机选择一个百分比进行训练、交叉验证和测试数据。 随机挑选数据集的常规和欺诈性数据的代码如下所示：

　　我们运行这个代码两次以获得训练和交叉验证数据。 剩下的是测试数据。 稍后我们会看到几个百分比的选择。

　　2.接下来，将需要μ（均值）和σ2（方差）的计算，因为它们对于获得新例子的概率至关重要。 代码如下所示：

　　3.如前所述，一旦我们使用高斯公式，具有均值和方差，就可以计算出概率值。 根据概率值，我们决定它是一个异常还是一个正常的例子。 将该值与某个阈值（ε）进行比较。如果它较低，那么我们将其标记为异常，如果更大，则将其标记为常规。 选择epsilon是至关重要的，因为具有小的价值会导致算法标记大量的虚假欺诈。 另一方面，大的价值，我们又会错过欺诈。 所以使用精确的交叉验证数据和召回选择非常好的的epsilon。

　　4.现在，我们准备在测试数据上评估我们的算法（我们也对交叉验证数据做了可选的评估）。

　　在执行算法之前，需要下载数据（因为GitHub的文件大小限制而没有打包），解压缩，然后把allData.csv复制粘贴到文件夹data / allData.csv。 文件位置是可配置的，以及文件名称。 该算法可以通过配置文件config / algorithm.properties中的数据和各种选项进行测试，如下所示：　

　　配置更改后，应用程序可以在Java IDE或Maven上运行：

　　根据机器和配置，应用程序可能需要一些时间（对我来说，这需要两分钟）完成。 此外，计算机可能会冻结，因为在某个时刻，Spark将CPU占用率降至100％。 此外，应用程序使用大量的内存（2-3 GB）。 可以看到打印在控制台上的结果或通过查看文件夹out /; 会有一个生成的文件* .txt与输出。 该算法是基于随机性的，所以你可以将其配置为运行多次，并且每次执行期望一个文件。

　　实验和结果

　　从这次的实验中，发现欺诈只适用于两种类型：TRANSFER和CASH_OUT。 在以前的文章中详细调查了TRANSFER。 我们取得了很高的比例：99.7％。

　　当仅为CASH_OUT类型运行而不跳过任何列/功能时，我们得到的结果很差：

　　我们只能找到约14％的这种欺诈行为。 以前，我们可以通过使特征看起来像高斯钟形曲线来改善很多，但不幸的是，这次并不是这样。

　　我们可以做的是看看功能，看看是否可以添加或跳过一些功能，因为功能cam=n引起混乱和噪音，而不是好处。查看数据源，我们有以下描述的欺诈，这可以帮助：

　　当大量的资金变现时，可能是欺诈。 慢慢地，我们开始删除不需要的功能，通过删除功能[1,2,3,7,8]或类型找到了很好的结果。 当兑现时，资金被占用的账户比目的地更重要，因为账户可能已经有钱了，而且看起来很正常，但是一个空的来源账户可能表示欺诈行为。 当离开目标帐户名称，它可能有助于欺诈性帐户名称。 结果如下所示：

　　这是一个巨大的进步。我们能够通过将所有类型组合在一起，从14％提高到82.77％。 此外，它不会带来不同的跳过功能的更好结果（随意尝试，因为并非所有这些都被探索）。 我只能通过跳过数量（2）得到一些结果，但这仍然不令人满意，因为大量的非欺诈活动被标记了（1,040,950）。

　　在这种情况下，为每种类型都运行算法可能会更好。 当一个可能的交易完成时，我们会反对它的类型。 通过这种方式，可以更好地检测到欺诈行为，因为转账有99.7％的利率，现金流有87％。 但是，对于现金流，我们可以说这个比率并不令人满意，也许还有其他的方法值得尝试，但这必须首先被调查，而且通常，直觉是错误的，还花费了很多时间。 由于隐私问题在金融应用中获取更多数据是困难的，所以宁愿在这里应用不同的算法。 当现金流的数据被绘制时，我们得到如下的视图：

　　该图显示问题在于大多数欺诈行为被包含在正常数据的中心，并且该算法努力检测它们。不过，我相信还有其他方法可以混合使用甚至能添加更多的功能。

　　Java流与Spark

　　我们可以配置算法（请参阅属性runsWith）在Spark或Java 8 Streams上运行以处理数据。如果要在集群上的多个远程节点上运行代码并将结果集合到请求的机器上，则Spark是一个很好的框架。在本文中，算法在本地执行，Spark将本地资源（如CPU数量）视为目标群集资源。另一方面，Java 8流很容易提供与collection.stram（）。parallel（）（当然，在本地运行的机器上）的并行性。因此，作为实验的一部分，Java 8流在一台机器上与Spark进行了比较。

　　结果表明，Java 8流在本地速度更快，即使不是太多。 Java = 111,927秒，Spark = 128,117秒。所以基本上，在运行所有数据时，流速要快16-25秒。请注意，每个人的电脑结果可能会有所不同。

　　由于Spark针对分布式计算进行了优化，与Java Streams相比，它在分区、任务等方面存在一些开销，只需要考虑本地机器，并且可以在那里进行优化。无论如何，都可以看到数据量在本地增加的差距。

　　对于少量的数据，Java 8 流更适合，但是对于大量的数据，Spark的缩放比例更好。也许值得尝试在AWS上运行的群集上配置Spark，而不是在本地。有关更多详细信息，请参阅处理相同精确算法的两个Java实现的代码，但具有不重要的小差异：Fraud Detection Algorithm Java Stream和Fraud Detection Algorithm Spark。