title: Spark: Counting co-occurrence - Algorithm for efficient multi-pass filtering of huge collections
body: <p>There is a table with two columns <code>books</code> and <code>readers</code> of these books, where <code>books</code> and <code>readers</code> are book and reader IDs, respectively :</p>
  
  <pre><code>   books readers
  1:     1      30
  2:     2      10
  3:     3      20
  4:     1      20
  5:     1      10
  6:     2      30
  </code></pre>
  
  <p>Record <code>book = 1, reader = 30</code> means that book with <code>id = 1</code> was read by user with <code>id = 30</code>.
  For each <strong>book pair</strong> I  need to count number of readers who read <em>both</em> of these books, with this algorithm:</p>
  
  <pre><code>for each book
    for each reader of the book
      for each other_book in books of the reader
        increment common_reader_count ((book, other_book), cnt)
  </code></pre>
  
  <p>The advantage of using this algorithm is that it requires a <strong>small number of  operations</strong> compared to counting all book combinations by two. </p>
  
  <p>To implement the above algorithm I organize this data in two groups : 1) keyed by book, an RDD containing readers of each book and 2) keyed by reader, an RDD containing books read by each reader, such as in the following program: </p>
  
  <pre><code>import org.apache.spark.SparkConf
  import org.apache.spark.SparkContext
  import org.apache.spark.SparkContext._
  import org.apache.log4j.Logger
  import org.apache.log4j.Level
  
  object Small {
  
    case class Book(book: Int, reader: Int)
    case class BookPair(book1: Int, book2: Int, cnt:Int)
  
    val recs = Array(
      Book(book = 1, reader = 30),
      Book(book = 2, reader = 10),
      Book(book = 3, reader = 20),
      Book(book = 1, reader = 20),
      Book(book = 1, reader = 10),
      Book(book = 2, reader = 30))
  
    def main(args: Array[String]) {
      Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
      Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)
      // set up environment
      val conf = new SparkConf()
        .setAppName("Test")
        .set("spark.executor.memory", "2g")
      val sc = new SparkContext(conf)
      val data = sc.parallelize(recs)
  
      val bookMap = data.map(r =&gt; (r.book, r))
      val bookGrps = bookMap.groupByKey
  
      val readerMap = data.map(r =&gt; (r.reader, r))
      val readerGrps = readerMap.groupByKey
  
      // *** Calculate book pairs
      // Iterate book groups 
      val allBookPairs = bookGrps.map(bookGrp =&gt; bookGrp match {
        case (book, recIter) =&gt;
          // Iterate user groups 
          recIter.toList.map(rec =&gt; {
            // Find readers for this book
            val aReader = rec.reader
            // Find all books (including this one) that this reader read
            val allReaderBooks = readerGrps.filter(readerGrp =&gt; readerGrp match {
              case (reader2, recIter2) =&gt; reader2 == aReader
            })
            val bookPairs = allReaderBooks.map(readerTuple =&gt; readerTuple match {
              case (reader3, recIter3) =&gt; recIter3.toList.map(rec =&gt; ((book, rec.book), 1))
            })
            bookPairs
          })
  
      })
      val x = allBookPairs.flatMap(identity)
      val y = x.map(rdd =&gt; rdd.first)
      val z = y.flatMap(identity)
      val p = z.reduceByKey((cnt1, cnt2) =&gt; cnt1 + cnt2)
      val result = p.map(bookPair =&gt; bookPair match {
        case((book1, book2),cnt) =&gt; BookPair(book1, book2, cnt)
      } )
  
      val resultCsv = result.map(pair =&gt; resultToStr(pair))
      resultCsv.saveAsTextFile("./result.csv")
    }
  
     def resultToStr(pair: BookPair): String = {
       val sep = "|"
      pair.book1 + sep + pair.book2 + sep + pair.cnt
    }
  }
  </code></pre>
  
  <p>This implemntation in fact results in <strong>the different, inefficient algorithm !</strong>:</p>
  
  <pre><code>for each book
    find each reader of the book scanning all readers every time!
      for each other_book in books of the reader
        increment common_reader_count ((book, other_book), cnt)
  </code></pre>
  
  <p>which contradicts the main goal of the discussed above algorithm because instead of reducing, it increases the number of operations. Finding user books requires filtering all users for every book. Thus number of operations ~ N * M where N - number of users and M - number of books.</p>
  
  <p>Questions:</p>
  
  <ol>
  <li>Is there any way to implement the original algorithm in Spark without filtering complete reader collection for every book?</li>
  <li>Any other algorithms to compute book pair counts efficiently?</li>
  <li>Also, when actually running this code I get <code>filter exception</code> which reason I can not figure out. Any ideas? </li>
  </ol>
  
  <p>Please, see exception log below:</p>
  
  <pre><code>15/05/29 18:24:05 WARN util.Utils: Your hostname, localhost.localdomain resolves to a loopback address: 127.0.0.1; using 10.0.2.15 instead (on interface eth0)
  15/05/29 18:24:05 WARN util.Utils: Set SPARK_LOCAL_IP if you need to bind to another address
  15/05/29 18:24:09 INFO slf4j.Slf4jLogger: Slf4jLogger started
  15/05/29 18:24:10 INFO Remoting: Starting remoting
  15/05/29 18:24:10 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriver@10.0.2.15:38910]
  15/05/29 18:24:10 INFO Remoting: Remoting now listens on addresses: [akka.tcp://sparkDriver@10.0.2.15:38910]
  15/05/29 18:24:12 ERROR executor.Executor: Exception in task 0.0 in stage 6.0 (TID 4)
  java.lang.NullPointerException
      at org.apache.spark.rdd.RDD.filter(RDD.scala:282)
      at Small$$anonfun$4$$anonfun$apply$1.apply(Small.scala:58)
      at Small$$anonfun$4$$anonfun$apply$1.apply(Small.scala:54)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
      at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:244)
      at scala.collection.immutable.List.foreach(List.scala:318)
      at scala.collection.TraversableLike$class.map(TraversableLike.scala:244)
      at scala.collection.AbstractTraversable.map(Traversable.scala:105)
      at Small$$anonfun$4.apply(Small.scala:54)
      at Small$$anonfun$4.apply(Small.scala:51)
      at scala.collection.Iterator$$anon$11.next(Iterator.scala:328)
      at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
      at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
      at scala.collection.Iterator$$anon$13.hasNext(Iterator.scala:371)
      at org.apache.spark.util.collection.ExternalAppendOnlyMap.insertAll(ExternalAppendOnlyMap.scala:137)
      at org.apache.spark.Aggregator.combineValuesByKey(Aggregator.scala:58)
      at org.apache.spark.shuffle.hash.HashShuffleWriter.write(HashShuffleWriter.scala:55)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:68)
      at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41)
      at org.apache.spark.scheduler.Task.run(Task.scala:54)
      at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)
      at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
      at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
      at java.lang.Thread.run(Thread.java:744)
  </code></pre>
  
  <p><em>Update:</em></p>
  
  <p>This code:</p>
  
  <pre><code>val df = sc.parallelize(Array((1,30),(2,10),(3,20),(1,10)(2,30))).toDF("books","readers")
  val results = df.join(
  df.select($"books" as "r_books", $"readers" as "r_readers"), 
  $"readers" === $"r_readers" and $"books" &lt; $"r_books"
  )
  .groupBy($"books", $"r_books")
  .agg($"books", $"r_books", count($"readers"))
  </code></pre>
  
  <p>Gives the following result:</p>
  
  <pre><code>books r_books COUNT(readers)
  1     2       2     
  </code></pre>
  
  <p>So <code>COUNT</code> here is a number of times two books (here 1 and 2) were read together (count of pairs).  </p>