title: Spark: How to join RDDs by time range
body: <p>I have a delicate Spark problem, where i just can't wrap my head around. </p>
  
  <p>We have two RDDs ( coming from Cassandra ). RDD1 contains <code>Actions</code> and RDD2 contains <code>Historic</code> data. Both have an id on which they can be matched/joined. But the problem is the two tables have an N:N relation ship. <code>Actions</code> contains multiple rows with the same id and so does <code>Historic</code>. Here are some example date from both tables.</p>
  
  <p><code>Actions</code> <em>time is actually a timestamp</em></p>
  
  <pre><code>id  |  time  | valueX
  1   |  12:05 | 500
  1   |  12:30 | 500
  2   |  12:30 | 125
  </code></pre>
  
  <p><code>Historic</code> <em>set_at is actually a timestamp</em></p>
  
  <pre><code>id  |  set_at| valueY
  1   |  11:00 | 400
  1   |  12:15 | 450
  2   |  12:20 | 50
  2   |  12:25 | 75
  </code></pre>
  
  <p>How can we join these two tables in a way, that we get a result like this</p>
  
  <pre><code>1   |  100  # 500 - 400 for Actions#1 with time 12:05 because Historic was in that time at 400
  1   |  50   # 500 - 450 for Actions#2 with time 12:30 because H. was in that time at 450
  2   |  50   # 125 - 75  for Actions#3 with time 12:30 because H. was in that time at 75
  </code></pre>
  
  <p>I can't come up with a good solution that feels right, without making a lot of iterations over huge datasets. I always have to think about making a range from the <code>Historic</code> set and then somehow check if the <code>Actions</code> fits in the range e.g (11:00 - 12:15) to make the calculation. But that seems to pretty slow to me. Is there any more efficient way to do that? Seems to me, that this kind of problem could be popular, but i couldn't find any hints on this yet. How would you solve this problem in spark?</p>
  
  <p>My current attempts so far ( in half way done code )</p>
  
  <pre><code>case class Historic(id: String, set_at: Long, valueY: Int)
  val historicRDD = sc.cassandraTable[Historic](...)
  
  historicRDD
  .map( row =&gt; ( row.id, row ) )
  .reduceByKey(...) 
  // transforming to another case which results in something like this; code not finished yet
  // (List((Range(0, 12:25), 400), (Range(12:25, NOW), 450)))
  
  // From here we could join with Actions
  // And then some .filter maybe to select the right Lists tuple
  </code></pre>