Spark上的DBSCAN：哪个实现

Question

我想在 Spark 上做一些 DBSCAN。我目前找到了 2 个实现：

• https://github.com/irvingc/dbscan-on-spark
• https://github.com/alitouka/spark_dbscan

我已经使用其 github 中给出的 sbt 配置测试了第一个，但是：

• jar 中的函数与 doc 或 github 上的源中的函数不同。比如我在jar中找不到train函数

• 我设法使用 fit 函数（在 jar 中找到）运行测试，但是 epsilon 的错误配置（有点大）使代码陷入无限循环。

代码：

val model = DBSCAN.fit(eps, minPoints, values, parallelism)

有人设法对第一个库做点什么吗？

有人测试过第二个吗？

Answer 1

请尝试ELKI。由于这是 Java，应该很容易从 Scala 调用。

ELKI 优化得非常好，通过索引可以扩展到相当大的数据集。

我们试图在我们的基准测试研究中包含其中一个 Spark 实现 - 但它内存不足（而且它是唯一一个内存不足的实现...... Spark 和 Mahout 的 k-means 也在其中最慢的）：

Hans-Peter Kriegel、Erich Schubert 和 Arthur Zimek。
The (black) art of runtime evaluation: Are we comparing algorithms or implementations?
在：知识和信息系统（KAIS）。 2016, 1-38

Neukirchen 教授在本技术报告中对 DBSCAN 的并行实现进行了基准测试：

赫尔穆特·纽基兴
Survey and Performance Evaluation of DBSCAN Spatial Clustering Implementations for Big Data and High-Performance Computing Paradigms

显然他让一些 Spark 实现工作，但注意到：

结果是毁灭性的：Apache Spark 的任何实现都无法与 HPC 实现相提并论。特别是在更大（但仍然相当小）的数据集上，它们中的大多数完全失败并且甚至不能提供正确的结果。

及更早：

在使用集群的所有可用核心运行任何“Spark DBSCAN”实现时，我们遇到了内存不足异常。

（另外，“Spark DBSCAN”在 928 个内核上耗时 2406 秒，ELKI 在 1 个内核上耗时 997 秒，用于较小的基准测试 - 其他 Spark 实现也表现不佳，特别是它没有返回正确的结果...)

“DBSCAN on Spark”没有崩溃，但返回完全错误 集群。

虽然“DBSCAN on Spark”完成得更快，但它已交付 完全错误的聚类结果。由于 DBSCAN 的运行时间长得无可救药 Spark 的实现已经具有最大内核数，我们没有执行 使用更少的内核进行测量。

您可以将 double[][] 数组包装为 ELKI 数据库：

// Adapter to load data from an existing array.
DatabaseConnection dbc = new ArrayAdapterDatabaseConnection(data);
// Create a database (which may contain multiple relations!)
Database db = new StaticArrayDatabase(dbc, null);
// Load the data into the database (do NOT forget to initialize...)
db.initialize();

// Squared Euclidean is faster than Euclidean.
Clustering<Model> c = new DBSCAN<NumberVector>(
  SquaredEuclideanDistanceFunction.STATIC, eps*eps, minpts).run(db);

for(Cluster<KMeansModel> clu : c.getAllClusters()) {
  // Process clusters
}

另请参阅：Java API example（特别是如何将 DBID 映射回行索引）。为了获得更好的性能，请将索引工厂（例如 new CoverTree.Factory(...)）作为第二个参数传递给 StaticArrayDatabase 构造函数。

Answer 2

我在我的项目中成功使用了第二个库（https://github.com/alitouka/spark_dbscan）。实际上，我无法使用它，如下所示：

libraryDependencies += "org.alitouka" % "spark_dbscan_2.10" % "0.0.4"

resolvers += "Aliaksei Litouka's repository" at "http://alitouka-public.s3-website-us-east-1.amazonaws.com/"

相反，我下载代码并将其更新为 spark 2.2.1 版本。此外，应该添加一些库。最后，将代码添加到我的项目中，它可以工作！

Answer 3

我测试了https://github.com/irvingc/dbscan-on-spark，可以说它消耗了大量内存。对于具有平滑分布的 400K 数据集，我使用了 -Xmx12084m，即使在这种情况下，它的工作时间也太长（> 20 分钟）。此外，它仅适用于 2D。我使用的是 maven 项目，而不是 sbt。

我还测试了第二个实现。这仍然是我找到的最好的。可惜作者从 2015 年开始就不支持了。Spark 的版本提升和版本冲突的解决确实花了一些时间。我需要它在 aws 上部署。

Answer 4

您还可以考虑使用提供 DBSCAN 实现的smile。您必须以最直接的方式将groupBy 与mapGroups 或flatMapGroups 结合使用，然后您将在那里运行dbscan。这是一个例子：

  import smile.clustering._

  val dataset: Array[Array[Double]] = Array(
    Array(100, 100),
    Array(101, 100),
    Array(100, 101),
    Array(100, 100),
    Array(101, 100),
    Array(100, 101),

    Array(0, 0),
    Array(1, 0),
    Array(1, 2),
    Array(1, 1)
  )

  val dbscanResult = dbscan(dataset, minPts = 3, radius = 5)
  println(dbscanResult)

  // output
  DBSCAN clusters of 10 data points:
  0     6 (60.0%)
  1     4 (40.0%)
  Noise     0 ( 0.0%)

如果您需要获得更多性能，您还可以编写用户定义的聚合函数 (UDAF)。

我在工作中使用这种方法对时间序列数据进行聚类，因此使用 Spark 的时间窗口函数进行分组，然后能够在每个窗口中执行 DBSCAN 允许我们并行化实现。

我受到以下article 的启发而这样做