title: Spark - Finding overlapping values Or A variation of finding mutual friends
body: <p>I have a problem which I'm trying to solve using Spark. I'm fairly new to Spark so I am not sure what would be the best way to design it. </p>
  
  <h2>Input:</h2>
  
  <pre><code>group1=user1,user2
  group2=user1,user2,user3
  group3=user2,user4
  group4=user1,user4
  group5=user3,user5
  group6=user3,user4,user5
  group7=user2,user4
  group8=user1,user5
  group9=user2,user4,user5
  group10=user4,user5
  </code></pre>
  
  <p>I want to find mutual group count between each pair of users. So for the above input, the output I'm expecting would be:</p>
  
  <h2>Output:</h2>
  
  <pre><code>1st user || 2nd user || mutual/intersection count || union count
  ------------------------------------------------------------
  user1        user2           2                       7
  user1        user3           1                       6
  user1        user4           1                       9
  user2        user4           3                       8
  </code></pre>
  
  <p>I think there are several ways to solve this problem, one of the solution could be:</p>
  
  <ul>
  <li>Create a key,value pair where key is user and value is group</li>
  <li>Group by key, then we will have a list of groups a user is belong to</li>
  <li>then find intersection/union between two groups</li>
  </ul>
  
  <p>Example:</p>
  
  <pre><code>(1st stage): Map
  group1=user1,user2 ==&gt;
            user1, group1
            user2, group1
  group2=user1,user2,user3 ==&gt;
            user1, group2
            user2, group2
            user3, group2
  ....
  ....
  ....
  
  
  (2nd stage): Reduce by key
  user1 -&gt; group1, group2, group4, group8
  user2 -&gt; group1, group2, group3, group7, group9
  </code></pre>
  
  <p>But my question is, what would be the best way to represent the count in the way I want after I reduce them by key? </p>
  
  <p>Is there any better way to handle this problem? The maximum number of user is constant and will not go more than 5000, so that's the maximum number of keys it will create. But input may contains several lines close to 1B rows. I don't think that would be a problem, please correct me if I'm wrong.</p>
  
  <h2>Update:</h2>
  
  <p>This is the piece of code I came up to solve this problem with my little knowledge of Spark (just started learning Spark last month):</p>
  
  <pre><code>def createPair(line: String): Array[(String, String)] = {
      val splits = line.split("=")
      val kuid = splits(0)
      splits(1).split(",").map { segment =&gt; (segment, kuid) }
  }
  
  
  val input = sc.textFile("input/test.log")
  val pair = input.flatMap { line =&gt; createPair(line) }
  
  val pairListDF = pair
    .aggregateByKey(scala.collection.mutable.ListBuffer.empty[String])(
      (kuidList, kuid) =&gt; { kuidList += kuid; kuidList },
      (kuidList1, kuidList2) =&gt; { kuidList1.appendAll(kuidList2); kuidList1 })
    .mapValues(_.toList).toDF().select($"_1".alias("user"), $"_2".alias("groups"))
  
  pairListDF.registerTempTable("table")
  
  sqlContext.udf.register("intersectCount", (list1: WrappedArray[String], list2: WrappedArray[String]) =&gt; list1.intersect(list2).size)
  sqlContext.udf.register("unionCount", (list1: WrappedArray[String], list2: WrappedArray[String]) =&gt; list1.union(list2).distinct.size)
  
  val populationDF = sqlContext.sql("SELECT t1.user AS user_first,"
    + "t2.user AS user_second,"
    + "intersectCount(t1.groups, t2.groups) AS intersect_count,"
    + "unionCount(t1.groups, t2.groups) AS union_count"
    + " FROM table t1 INNER JOIN table t2"
    + " ON t1.user &lt; t2.user"
    + " ORDER BY user_first,user_second")
  </code></pre>
  
  <h2>Output:</h2>
  
  <pre><code>+----------+-----------+---------------+-----------+
  |user_first|user_second|intersect_count|union_count|
  +----------+-----------+---------------+-----------+
  |     user1|      user2|              2|          7|
  |     user1|      user3|              1|          6|
  |     user1|      user4|              1|          9|
  |     user1|      user5|              1|          8|
  |     user2|      user3|              1|          7|
  |     user2|      user4|              3|          8|
  |     user2|      user5|              1|          9|
  |     user3|      user4|              1|          8|
  |     user3|      user5|              2|          6|
  |     user4|      user5|              3|          8|
  +----------+-----------+---------------+-----------+
  </code></pre>
  
  <p>Would love to get some feedbacks about my code and things I'm missing. Please feel free to criticize my code as I just started learning Spark. Thanks again @axiom for your answer, much smaller and nicer solution than I expected.</p>