Dynamic Set Algebra on Spark

Question

Betrachten Sie das folgende Problem. Gegeben:

1. Eine Sammlung von Sätzen
2. Ein Boolescher Ausdruck auf ihnen, die dynamisch

Return die resultierende Menge empfangen wird.

Does Funken haben keine effizienten Algorithmen oder Bibliotheken dieses allgemeine Problem zu lösen? Hier

ist ein Spielzeug Beispiel das Problem konzeptionell zu illustrieren:

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

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

ich nach einem Weg suchen eine allgemeine solve_expression so dass der Implementierung, in dem obigen Beispiel:

output = solve_expression(expression, collection) 

Ergebnisse in:

Set("A2", "A5") 

---

ich mit Sätzen mit Millionen von Artikeln arbeitete, und Booleschen Ausdrücken, die als Strings kommen. Was wichtig ist, ist, dass jedes Atom in dem Ausdruck (z. B. "X" und "Y" oben) Sätze sind. Die Ausdrücke und Mengen sind dynamische (die Operationen können nicht hart-codiert werden, da wir sie als eine Eingabe erhalten und wir nicht wissen, was sie vorher sind).

Ich bin flexibel mit der Darstellung des Problems. Die tatsächlichen Sätze können vom Typ Set, z.B. Halten von Zeichenfolgen (z. B. "A1", "A2"), codiert als binäre Vektoren oder irgendetwas anderes, das dies für Spark zugänglich macht.

Does Funken haben keine Bibliotheken zu analysieren und allgemeine Boolesche Ausdrücke auf Sätze lösen?

Answer 1

In Ordnung. Angenommen, Sie möchten dies in Spark tun. Da es sich hierbei um riesige Mengen handelt, nehmen wir an, dass sie sich noch nicht im Speicher befinden. Sie befinden sich jeweils in einer Datei - jede Zeile in einer Datei bezeichnet einen Eintrag in der Menge.

Wir Sets repräsentieren mit RDD s - Standardweg Spark Daten zu speichern.

Unter Verwendung dieses Parsers (angepasst und fixiert von 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 
} 

Beachten Sie die folgende spark-shell Interaktion nach dem oben in die Schale eingefügt zu haben (Ich habe einige der Antworten der Kürze halber weggelassen):

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

Jetzt kann ich Abfrage mit verschiedenen Ausdrücken nennen. Beachten Sie, dass ich hier collect verwende, um die Auswertung auszulösen (also sehen wir, was in der Menge ist), aber wenn die Mengen wirklich groß sind, würden Sie normalerweise einfach die RDD behalten (und sie auf Festplatte speichern).

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

Obwohl ich nicht zu ihrer Umsetzung störte, soll eingestellt Unterschied eine einzeilige Zugabe (sehr ähnlich union und inter) sein. Ich denke, Satzkomplemente sind eine schlechte Idee ... sie ergeben nicht immer einen Sinn (was ist die Ergänzung des leeren Satzes, wie stellst du ihn dar?).