title: Why does df.limit keep changing in Pyspark?
body: <p>I'm creating a data sample from some dataframe <code>df</code> with</p>
  
  <pre><code>rdd = df.limit(10000).rdd
  </code></pre>
  
  <p>This operation takes quite some time (why actually? can it not short-cut after 10000 rows?), so I assume I have a new RDD now.</p>
  
  <p>However, when I now work on <code>rdd</code>, it is different rows every time I access it. As if it resamples over again. Caching the RDD helps a bit, but surely that's not save?</p>
  
  <p>What is the reason behind it?</p>
  
  <p>Update: Here is a reproduction on Spark 1.5.2</p>
  
  <pre><code>from operator import add
  from pyspark.sql import Row
  rdd=sc.parallelize([Row(i=i) for i in range(1000000)],100)
  rdd1=rdd.toDF().limit(1000).rdd
  for _ in range(3):
      print(rdd1.map(lambda row:row.i).reduce(add))
  </code></pre>
  
  <p>The output is</p>
  
  <pre><code>499500
  19955500
  49651500
  </code></pre>
  
  <p>I'm surprised that <code>.rdd</code> doesn't fix the data.</p>
  
  <p>EDIT:
  To show that it get's more tricky than the re-execution issue, here is a single action which produces incorrect results on Spark 2.0.0.2.5.0</p>
  
  <pre><code>from pyspark.sql import Row
  rdd=sc.parallelize([Row(i=i) for i in range(1000000)],200)
  rdd1=rdd.toDF().limit(12345).rdd
  rdd2=rdd1.map(lambda x:(x,x))
  rdd2.join(rdd2).count()
  # result is 10240 despite doing a self-join
  </code></pre>
  
  <p>Basically, whenever you use <code>limit</code> your results might be potentially wrong. I don't mean "just one of many samples", but really incorrect (since in the case the result should always be 12345).</p>