Spark上的动态集代数

Question

考虑以下问题.鉴于:

1. 集合的集合
2. 它们的动态接收的布尔表达式

返回结果集.

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")

---

我正在处理具有数百万项的集合,以及作为字符串的布尔表达式.重要的是表达式中的每个原子(例如上面的"X"和"Y")都是集合.表达式和集合是动态的(操作不能硬编码,因为我们将它们作为输入接收,我们事先不知道它们是什么).

我对问题的表现很灵活.实际集合可以是类型Set,例如保持字符串(例如"A1","A2"),编码为二进制向量,或任何使其适合Spark的其他集合.

Spark有没有任何库来解析和解决集合上的一般布尔表达式？

Answer 1

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

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

使用此解析器（从此处进行调整和修复）

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
}

将上述内容粘贴到 shell 后观察以下spark-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")

虽然我没有费心去实现它，但 set Difference 应该是一行加法（与unionand非常相似inter）。我认为集合补码是一个坏主意......它们并不总是有意义（空集的补码是什么，你如何表示它？）。