diff options
author | Shivaram Venkataraman <shivaram@eecs.berkeley.edu> | 2013-06-05 14:28:38 -0700 |
---|---|---|
committer | Shivaram Venkataraman <shivaram@eecs.berkeley.edu> | 2013-06-05 14:28:38 -0700 |
commit | c851957fe4798d5dfb8deba7bf79a035a0543c74 (patch) | |
tree | 193e29ab284cf1a0d2c61413cdebeca90767c247 /core/src/test/scala | |
parent | 96943a1cc054d7cf80eb8d3dfc7fb19ce48d3c0a (diff) | |
download | spark-c851957fe4798d5dfb8deba7bf79a035a0543c74.tar.gz spark-c851957fe4798d5dfb8deba7bf79a035a0543c74.tar.bz2 spark-c851957fe4798d5dfb8deba7bf79a035a0543c74.zip |
Don't write zero block files with java serializer
Diffstat (limited to 'core/src/test/scala')
-rw-r--r-- | core/src/test/scala/spark/ShuffleSuite.scala | 26 |
1 files changed, 26 insertions, 0 deletions
diff --git a/core/src/test/scala/spark/ShuffleSuite.scala b/core/src/test/scala/spark/ShuffleSuite.scala index b967016cf7..33b02fff80 100644 --- a/core/src/test/scala/spark/ShuffleSuite.scala +++ b/core/src/test/scala/spark/ShuffleSuite.scala @@ -367,6 +367,32 @@ class ShuffleSuite extends FunSuite with ShouldMatchers with LocalSparkContext { assert(nonEmptyBlocks.size <= 4) } + test("zero sized blocks without kryo") { + // Use a local cluster with 2 processes to make sure there are both local and remote blocks + sc = new SparkContext("local-cluster[2,1,512]", "test") + + // 10 partitions from 4 keys + val NUM_BLOCKS = 10 + val a = sc.parallelize(1 to 4, NUM_BLOCKS) + val b = a.map(x => (x, x*2)) + + // NOTE: The default Java serializer doesn't create zero-sized blocks. + // So, use Kryo + val c = new ShuffledRDD(b, new HashPartitioner(10)) + + val shuffleId = c.dependencies.head.asInstanceOf[ShuffleDependency[Int, Int]].shuffleId + assert(c.count === 4) + + val blockSizes = (0 until NUM_BLOCKS).flatMap { id => + val statuses = SparkEnv.get.mapOutputTracker.getServerStatuses(shuffleId, id) + statuses.map(x => x._2) + } + val nonEmptyBlocks = blockSizes.filter(x => x > 0) + + // We should have at most 4 non-zero sized partitions + assert(nonEmptyBlocks.size <= 4) + } + } object ShuffleSuite { |