fix seed distribution and add some tests for rdd.sample

Author: mengxr

python/pyspark/rdd.py

@@ -317,8 +317,23 @@ def sample(self, withReplacement, fraction, seed=None):
         Return a sampled subset of this RDD (relies on numpy and falls back
         on default random generator if numpy is unavailable).
 
-        >>> sc.parallelize(range(0, 100)).sample(False, 0.1, 2).collect() #doctest: +SKIP
-        [2, 3, 20, 21, 24, 41, 42, 66, 67, 89, 90, 98]
+        >>> rdd = sc.parallelize(range(0, 100), 4)
+        >>> wo = rdd.sample(False, 0.1, 2).collect()
+        >>> wo_dup = rdd.sample(False, 0.1, 2).collect()
+        >>> set(wo) == set(wo_dup)
+        True
+        >>> wr = rdd.sample(True, 0.2, 5).collect()
+        >>> wr_dup = rdd.sample(True, 0.2, 5).collect()
+        >>> set(wr) == set(wr_dup)
+        True
+        >>> wo_s10 = rdd.sample(False, 0.3, 10).collect()
+        >>> wo_s20 = rdd.sample(False, 0.3, 20).collect()
+        >>> set(wo_s10) != set(wo_s20)
+        True
+        >>> wr_s11 = rdd.sample(True, 0.4, 11).collect()
+        >>> wr_s21 = rdd.sample(True, 0.4, 21).collect()
+        >>> set(wr_s11) != set(wr_s21)
+        True
         """
         assert fraction >= 0.0, "Negative fraction value: %s" % fraction
         return self.mapPartitionsWithIndex(RDDSampler(withReplacement, fraction, seed).func, True)

python/pyspark/rddsampler.py

@@ -40,14 +40,13 @@ def __init__(self, withReplacement, seed=None):
     def initRandomGenerator(self, split):
         if self._use_numpy:
             import numpy
-            self._random = numpy.random.RandomState(self._seed)
+            self._random = numpy.random.RandomState(self._seed ^ split)
         else:
-            self._random = random.Random(self._seed)
+            self._random = random.Random(self._seed ^ split)
 
-        for _ in range(0, split):
-            # discard the next few values in the sequence to have a
-            # different seed for the different splits
-            self._random.randint(0, 2 ** 32 - 1)
+        # mixing because the initial seeds are close to each other
+        for _ in xrange(10):
+            self._random.randint(0, 1)
 
         self._split = split
         self._rand_initialized = True