title: Join and map 2 RDDs conditionally
body: <p>I have 2 RDDs which need to be joined</p>
  
  <pre><code>val rdd1 = RDD[(v_id, inputObject1)]
  </code></pre>
  
  <p>where v_id is unique id</p>
  
  <p>and inputObject1 has following fields</p>
  
  <pre><code>g_id, p_id, timestamp=t1
  </code></pre>
  
  <p>Now i have another RDD</p>
  
  <pre><code>val rdd2 = RDD[(g_id, inputObject2)]
  </code></pre>
  
  <p>where inputObject2 has following fields</p>
  
  <pre><code>p_id, timestamp=t2, e_id
  </code></pre>
  
  <p>Now i want to join these 2 RDDs on below condition</p>
  
  <ul>
  <li>If g_id and p_id is same and |t1-t2| &lt; 30 minutes</li>
  <li>Else if g_id is same and |t1 - t2| &lt; 30 minutes</li>
  </ul>
  
  <p>So second condition is fallback if first condition is not met. My final output should be this</p>
  
  <pre><code>val resuldRDD = RDD[(v_id, inputObject11)]
  </code></pre>
  
  <p>Where inputObject11 = inputObject1 + adding e_id from second RDD if conditions are met.</p>
  
  <p>So fields will be</p>
  
  <pre><code>g_id, p_id, e_id, timestamp=t1
  </code></pre>