title: Spark - Nested RDD Operation
body: <p>I have two RDDs say</p>
  
  <pre><code>   rdd1 = 
  id            | created     | destroyed | price   
  1            | 1            | 2            | 10        
  2            | 1            | 5            | 11       
  3            | 2            | 3            | 11        
  4            | 3            | 4            | 12        
  5            | 3            | 5            | 11       
  
  rdd2 =
  
  [1,2,3,4,5] # lets call these value as timestamps (ts)
  </code></pre>
  
  <p>rdd2 is basically generated using range(intial_value, end_value, interval). The params here can vary. The size can be same or different to rdd1. The idea is to fetch records from rdd1 into rdd2 based on the values of rdd2 using a filtering criertia(records from rdd1 can repeat while fetching as you can see in output)</p>
  
  <p>filtering criteria rdd1.created &lt;= ts &lt; rdd1.destroyed)</p>
  
  <p>Expected output:</p>
  
  <pre><code>ts             | prices  
  1              | 10,11       # i.e. for ids 1,2 of rdd1      
  2              | 11,11       # ids 2,3
  3              | 11,12,11    # ids 2,4,5 
  4              | 11,11       # ids 2,5
  </code></pre>
  
  <p>Now I want to filter RDD1 based on some condition which <strong>uses the keys of RDD2. (described above)</strong> And returns the results which joins the keys of RDD2 and filtered results of RDD1</p>
  
  <p>So I do:</p>
  
  <pre><code>rdd2.map(lambda x : somefilterfunction(x, rdd1))  
  
  def somefilterfunction(x, rdd1):
      filtered_rdd1 = rdd1.filter(rdd1[1] &lt;= x).filter(rdd1[2] &gt; x)
      prices = filtered_rdd1.map(lambda x : x[3])
      res = prices.collect()
      return (x, list(res))
  </code></pre>
  
  <p>And I get:</p>
  
  <blockquote>
    <p>Exception: It appears that you are attempting to broadcast an RDD or
    reference an RDD from an action or transformation. RDD transformations
    and actions can only be invoked by the driver, not inside of other
    transformations; for example, rdd1.map(lambda x: rdd2.values.count() *
    x) is invalid because the values transformation and count action
    cannot be performed inside of the rdd1.map transformation. For more
    information, see SPARK-5063.</p>
  </blockquote>
  
  <p>I tried using groupBy , but since here elements of rdd1 can be repeated again and again as compared to grouping which I understand would club each element of rdd1 in some particular slot just once. </p>
  
  <p>The only way is now to use a normal for loop and do the filtering and join everything in the end.</p>
  
  <p>Any Suggestions?</p>