title: PySpark: How do I send the RDD to a map function with dataFrame input in spark
body: <p>I am trying to used RDD in pyspark. Starting with the simple example estimating  the pi where:</p>
  
  <pre><code>from pyspark import SparkContext, SparkConf
  
  appName = "tutApp"
  master = "local"
  NUM_SAMPLE = 100
  NUM_PARALLIZE = xrange(0, NUM_SAMPLE)
  
  import random
  def sample(p):
      x, y = random.random(), random.random()
      return 1 if x*x + y*y &lt; 1 else 0
  
  count = sc.parallelize(NUM_PARALLIZE).map(sample).reduce(lambda a, b: a+b)
  t = (4.0 * count/NUM_SAMPLE)
  print "----- Pi is roughly %", t
  </code></pre>
  
  <p>I want to use the same approach, where the map function in my case (APR),  getting input, which is the data frame instead of x,y in the sample function. I am using pandas and it seems that spark has difficulty for distributing tasks by map function in my case.</p>
  
  <pre><code>sc.parallelize(PrepAPR(0,1)).count() 
  </code></pre>
  
  <p>gives me the number of nodes and this part is working. </p>
  
  <p>PrepAPR is a class where I am returning the tuple in a lazy way that can be used as an input for the map function. Having the function sample, I want to do transformation as </p>
  
  <pre><code>.map(lambda t: APR (t)) and t is a tuple of (df,a,b,c). 
  </code></pre>
  
  <p>I am getting this error </p>
  
  <pre><code>SystemError: error return without exception set.
  </code></pre>
  
  <p>I appreciate if some one answer my question. </p>
  
  <p>"""APR SPark"""</p>
  
  <pre><code>from pyspark import SparkContext, SparkConf
  
  class PrepAPR:
      def __init__(self, low, high):
          self.current = low
          self.high = high
  
      def __iter__(self):
          return self
  
      def prep_data_for_turbine(self):
          turbines=all_turbines[self.current:self.current+1]
          return turbines
  
      def next(self): # Python 3: def __next__(self)
          if self.current &gt; self.high:
              raise StopIteration
          else:
  #             tpl =  self.current 
              tpl = self.prep_data_for_turbine()
              self.current += 1
              return tpl
  def sami(t):
      model= MODEL
      df= dFrames["Data"]
      features=FEATURES
      turbines=all_turbines[t:t+1]
      return actual_pred_resid (df, model, features, turbines)
  
  
  # print sc.parallelize(PrepAPR(0,1)).count()
  APR_df = sc.parallelize([1,2]).map(lambda t: sami(t)).count()
  #print APR_df.collect()
  </code></pre>