scala - Spark 上的动态集合代数

Question

考虑以下问题。鉴于:

一套集合

动态接收的 bool 表达式

返回结果集。

Spark 是否有任何有效的算法或库来解决这个一般问题？

这是一个 玩具 从概念上说明问题的示例:

val X  = Set("A1", "A2", "A3", "A4")
val Y  = Set("A2", "A4", "A5")

val collection = Set(X, Y)
val expression = "X and Y"

我正在寻找一种实现通用的方法 solve_expression所以，在上面的例子中:

output = solve_expression(expression, collection)

结果是:

Set("A2", "A5")

我正在处理包含数百万个项目的集合，以及作为字符串出现的 bool 表达式。重要的是表达式中的每个原子(例如上面的“X”和“Y”)都是集合。表达式和集合是动态的(操作不能硬编码，因为我们将它们作为输入接收，而我们事先不知道它们是什么)。

我对问题的表述很灵活。实际集合的类型可以是 Set ，例如保存字符串(例如“A1”，“A2”)，编码为二进制向量，或任何其他使这适合 Spark 的东西。

Spark 是否有任何库到 解析 和 解决 集合上的通用 bool 表达式？

Answer 1

好吧。假设您想在 Spark 中执行此操作。此外，由于这些是巨大的集合，让我们假设它们还没有在内存中，它们每个都在一个文件中 - 文件中的每一行表示集合中的一个条目。

我们将用 RDD 表示集合s - Spark 存储数据的标准方式。

使用这个解析器(改编自 here )

import scala.util.parsing.combinator.JavaTokenParsers
import org.apache.spark.rdd.RDD

case class Query[T](setMap: Map[String, RDD[T]]) extends JavaTokenParsers {
  private lazy val expr: Parser[RDD[T]]
    = term ~ rep("union" ~ term) ^^ { case f1 ~ fs => (f1 /: fs)(_ union _._2) }
  private lazy val term: Parser[RDD[T]]
    = fact ~ rep("inter" ~ fact) ^^ { case f1 ~ fs => (f1 /: fs)(_ intersection _._2) }
  private lazy val fact: Parser[RDD[T]]
    = vari | ("(" ~ expr ~ ")" ^^ { case "(" ~ exp ~ ")" => exp })
  private lazy val vari: Parser[RDD[T]]
    = setMap.keysIterator.map(Parser(_)).reduceLeft(_ | _) ^^ setMap

  def apply(expression: String) = this.parseAll(expr, expression).get.distinct
}

遵守以下内容 spark-shell将上述内容粘贴到 shell 后进行交互(为简洁起见，我省略了一些回复):

> val x = sc.textFile("X.txt").cache \\ contains "1\n2\n3\n4\n5"
> val y = sc.textFile("Y.txt").cache \\ contains "3\n4\n5\n6\n7"
> val z = sc.textFile("Z.txt").cache \\ contains "3\n9\n\10"
> val sets = Map("x" -> x, "y" -> y, "z" -> z)
> val query = Query[Int](sets)

现在，我可以使用不同的表达式调用查询。请注意，这里我使用的是 collect触发评估(所以我们可以看到集合里面有什么)，但是如果集合真的很大，你通常只保留 RDD按原样(并将其保存到磁盘)。

> query("a union b").collect
res: Array[Int] = Array("1", "2", "3", "4", "5", "6", "7")
> query("a inter b").collect
res: Array[Int] = Array("3", "4", "5")
> query("a inter b union ((a inter b) union a)").collect
res: Array[Int] = Array("1", "2", "3", "4", "5")
> query("c union a inter b").collect
res: Array[Int] = Array("3", "4", "5", "9", "10")
> query("(c union a) inter b").collect
res: Array[Int] = Array("3", "4", "5")

虽然我没有费心去实现它，但集差应该是一行加法(非常类似于 union 和 inter )。我认为集合补是一个坏主意......它们并不总是有意义(空集的补是什么，你如何表示它？)。