Add -1 delimiter

Author: Feynman Liang

mllib/src/main/scala/org/apache/spark/mllib/fpm/LocalPrefixSpan.scala

@@ -25,7 +25,7 @@ import org.apache.spark.Logging
  * Calculate all patterns of a projected database in local.
  */
 private[fpm] object LocalPrefixSpan extends Logging with Serializable {
-
+  import PrefixSpan._
   /**
    * Calculate all patterns of a projected database.
    * @param minCount minimum count
@@ -43,7 +43,9 @@ private[fpm] object LocalPrefixSpan extends Logging with Serializable {
       database: Iterable[Array[Int]]): Iterator[(List[Int], Long)] = {
     if (prefixes.length == maxPatternLength || database.isEmpty) return Iterator.empty
     val frequentItemAndCounts = getFreqItemAndCounts(minCount, database)
-    val filteredDatabase = database.map(x => x.filter(frequentItemAndCounts.contains))
+    val filteredDatabase = database.map { suffix =>
+      suffix.filter(item => item == DELIMITER || frequentItemAndCounts.contains(item))
+    }
     frequentItemAndCounts.iterator.flatMap { case (item, count) =>
       val newPrefixes = item :: prefixes
       val newProjected = project(filteredDatabase, item)
@@ -63,7 +65,8 @@ private[fpm] object LocalPrefixSpan extends Logging with Serializable {
     if (index == -1) {
       Array()
     } else {
-      sequence.drop(index + 1)
+      // drop until we get to the next delimiter (or end of sequence)
+      sequence.drop(index).dropWhile(_ != DELIMITER).drop(1)
     }
   }
 
@@ -89,6 +92,6 @@ private[fpm] object LocalPrefixSpan extends Logging with Serializable {
         counts(item) += 1L
       }
     }
-    counts.filter(_._2 >= minCount)
+    counts.filter { case (item, count) => (count >= minCount) && (item != DELIMITER) }
   }
 }

mllib/src/main/scala/org/apache/spark/mllib/fpm/PrefixSpan.scala

@@ -44,6 +44,7 @@ import org.apache.spark.storage.StorageLevel
 class PrefixSpan private (
     private var minSupport: Double,
     private var maxPatternLength: Int) extends Logging with Serializable {
+  import PrefixSpan._
 
   /**
    * The maximum number of items allowed in a projected database before local processing. If a
@@ -90,12 +91,14 @@ class PrefixSpan private (
 
   /**
    * Find the complete set of sequential patterns in the input sequences.
-   * @param sequences input data set, contains a set of sequences,
-   *                  a sequence is an ordered list of elements.
+   * @param sequences a dataset of sequences. Items in a sequence are represented by non-negative
+   *                  integers and delimited by [[DELIMITER]]. Non-temporal sequences
+   *                  are supported by placing more than one item between delimiters.
    * @return a set of sequential pattern pairs,
    *         the key of pair is pattern (a list of elements),
    *         the value of pair is the pattern's count.
    */
+  // TODO: generalize to arbitrary item-types and use mapping to Ints for internal algorithm
   def run(sequences: RDD[Array[Int]]): RDD[(Array[Int], Long)] = {
     val sc = sequences.sparkContext
 
@@ -110,14 +113,14 @@ class PrefixSpan private (
     val freqItemCounts = sequences
       .flatMap(seq => seq.distinct.map(item => (item, 1L)))
       .reduceByKey(_ + _)
-      .filter(_._2 >= minCount)
+      .filter { case (item, count) => (count >= minCount) && (item != DELIMITER) }
       .collect()
 
     // Pairs of (length 1 prefix, suffix consisting of frequent items)
     val itemSuffixPairs = {
       val freqItems = freqItemCounts.map(_._1).toSet
       sequences.flatMap { seq =>
-        val filteredSeq = seq.filter(freqItems.contains(_))
+        val filteredSeq = seq.filter(item => item == DELIMITER || freqItems.contains(item))
         freqItems.flatMap { item =>
           val candidateSuffix = LocalPrefixSpan.getSuffix(item, filteredSeq)
           candidateSuffix match {
@@ -127,7 +130,6 @@ class PrefixSpan private (
         }
       }
     }
-
     // Accumulator for the computed results to be returned, initialized to the frequent items (i.e.
     // frequent length-one prefixes)
     var resultsAccumulator = freqItemCounts.map(x => (List(x._1), x._2))
@@ -197,7 +199,7 @@ class PrefixSpan private (
     val prefixItemPairAndCounts = prefixSuffixPairs
       .flatMap { case (prefix, suffix) => suffix.distinct.map(y => ((prefix, y), 1L)) }
       .reduceByKey(_ + _)
-      .filter(_._2 >= minCount)
+      .filter { case (item, count) => (count >= minCount) && (item != DELIMITER) }
 
     // Map from prefix to set of possible next items from suffix
     val prefixToNextItems = prefixItemPairAndCounts
@@ -247,3 +249,7 @@ class PrefixSpan private (
     }
   }
 }
+
+object PrefixSpan {
+  val DELIMITER = -1
+}

mllib/src/test/scala/org/apache/spark/mllib/fpm/PrefixSpanSuite.scala

@@ -40,7 +40,7 @@ class PrefixSpanSuite extends SparkFunSuite with MLlibTestSparkContext {
       Array(2, 4, 1),
       Array(3, 1, 3, 4, 5),
       Array(3, 4, 4, 3),
-      Array(6, 5, 3))
+      Array(6, 5, 3)).map(insertDelimiter)
 
     val rdd = sc.parallelize(sequences, 2).cache()
 
@@ -69,7 +69,7 @@ class PrefixSpanSuite extends SparkFunSuite with MLlibTestSparkContext {
       (Array(4, 5), 2L),
       (Array(5), 3L)
     )
-    assert(compareResults(expectedValue1, result1.collect()))
+    compareResults(expectedValue1, result1.collect())
 
     prefixspan.setMinSupport(0.5).setMaxPatternLength(50)
     val result2 = prefixspan.run(rdd)
@@ -80,7 +80,7 @@ class PrefixSpanSuite extends SparkFunSuite with MLlibTestSparkContext {
       (Array(4), 4L),
       (Array(5), 3L)
     )
-    assert(compareResults(expectedValue2, result2.collect()))
+    compareResults(expectedValue2, result2.collect())
 
     prefixspan.setMinSupport(0.33).setMaxPatternLength(2)
     val result3 = prefixspan.run(rdd)
@@ -100,14 +100,20 @@ class PrefixSpanSuite extends SparkFunSuite with MLlibTestSparkContext {
       (Array(4, 5), 2L),
       (Array(5), 3L)
     )
-    assert(compareResults(expectedValue3, result3.collect()))
+    compareResults(expectedValue3, result3.collect())
   }
 
   private def compareResults(
     expectedValue: Array[(Array[Int], Long)],
-    actualValue: Array[(Array[Int], Long)]): Boolean = {
-    expectedValue.map(x => (x._1.toSeq, x._2)).toSet ==
-      actualValue.map(x => (x._1.toSeq, x._2)).toSet
+    actualValue: Array[(Array[Int], Long)]): Unit = {
+    assert(expectedValue.map(x => (x._1.toSeq, x._2)).toSet ===
+      actualValue.map(x => (x._1.toSeq, x._2)).toSet)
+  }
+
+  private def insertDelimiter(sequence: Array[Int]): Array[Int] = {
+    sequence.zip(Seq.fill(sequence.length)(PrefixSpan.DELIMITER)).map { case (a, b) =>
+      List(a, b)
+    }.flatten
   }
 
 }