Why does df.limit keep changing in Pyspark?
<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>