only convert to list if __len__ not available, set number of partitions in test

Author: aray

python/pyspark/serializers.py

@@ -343,8 +343,8 @@ def _load_stream_without_unbatching(self, stream):
         key_batch_stream = self.key_ser._load_stream_without_unbatching(stream)
         val_batch_stream = self.val_ser._load_stream_without_unbatching(stream)
         for (key_batch, val_batch) in zip(key_batch_stream, val_batch_stream):
-            key_batch = list(key_batch)
-            val_batch = list(val_batch)
+            key_batch = key_batch if hasattr(key_batch, '__len__') else list(key_batch)
+            val_batch = val_batch if hasattr(val_batch, '__len__') else list(val_batch)
             if len(key_batch) != len(val_batch):
                 raise ValueError("Can not deserialize PairRDD with different number of items"
                                  " in batches: (%d, %d)" % (len(key_batch), len(val_batch)))

python/pyspark/tests.py

@@ -646,7 +646,7 @@ def test_cartesian_chaining(self):
 
     def test_zip_chaining(self):
         # Tests for SPARK-21985
-        rdd = self.sc.parallelize('abc')
+        rdd = self.sc.parallelize('abc',2)
         self.assertSetEqual(
             set(rdd.zip(rdd).zip(rdd).collect()),
             set([((x, x), x) for x in 'abc'])