Fixed bug in treeAggregate using fold

jkbradley · jkbradley · commit 02d107aaa493 · 2016-04-07T17:51:50.000-07:00
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -1089,9 +1089,8 @@ abstract class RDD[T: ClassTag](
           (i, iter) => iter.map((i % curNumPartitions, _))
         }.foldByKey(zeroValue, new HashPartitioner(curNumPartitions))(cleanCombOp).values
       }
-      //partiallyAggregated.reduce(cleanCombOp)
-      // This fails:
-      partiallyAggregated.fold(zeroValue)(cleanCombOp)
+      val copiedZeroValue = Utils.clone(zeroValue, sc.env.closureSerializer.newInstance())
+      partiallyAggregated.fold(copiedZeroValue)(cleanCombOp)
     }
   }
 

Original file line number	Diff line number	Diff line change
`@@ -1089,9 +1089,8 @@ abstract class RDD[T: ClassTag](`
`1089`	`1089`	`(i, iter) => iter.map((i % curNumPartitions, _))`
`1090`	`1090`	`}.foldByKey(zeroValue, new HashPartitioner(curNumPartitions))(cleanCombOp).values`
`1091`	`1091`	`}`
`1092`		`- //partiallyAggregated.reduce(cleanCombOp)`
`1093`		`- // This fails:`
`1094`		`- partiallyAggregated.fold(zeroValue)(cleanCombOp)`
	`1092`	`+ val copiedZeroValue = Utils.clone(zeroValue, sc.env.closureSerializer.newInstance())`
	`1093`	`+ partiallyAggregated.fold(copiedZeroValue)(cleanCombOp)`
`1095`	`1094`	`}`
`1096`	`1095`	`}`
`1097`	`1096`