diff options
author | Tathagata Das <tathagata.das1565@gmail.com> | 2013-01-14 23:06:40 -0800 |
---|---|---|
committer | Tathagata Das <tathagata.das1565@gmail.com> | 2013-01-14 23:06:40 -0800 |
commit | eded21925ab549330d0337138fa1f81ae192e3e9 (patch) | |
tree | c3cd6105aed3d77a8ae2c4dc64f2696aef9801ed /core | |
parent | 82b0cc90cad4eb0e85a7ff4e14a90afec258a9a1 (diff) | |
parent | 1638fcb0dce296da22ffc90127d5148a8fab745e (diff) | |
download | spark-eded21925ab549330d0337138fa1f81ae192e3e9.tar.gz spark-eded21925ab549330d0337138fa1f81ae192e3e9.tar.bz2 spark-eded21925ab549330d0337138fa1f81ae192e3e9.zip |
Merge pull request #375 from tdas/streaming
Important bug fixes
Diffstat (limited to 'core')
-rw-r--r-- | core/src/main/scala/spark/rdd/CheckpointRDD.scala | 6 |
1 files changed, 3 insertions, 3 deletions
diff --git a/core/src/main/scala/spark/rdd/CheckpointRDD.scala b/core/src/main/scala/spark/rdd/CheckpointRDD.scala index 86c63ca2f4..6f00f6ac73 100644 --- a/core/src/main/scala/spark/rdd/CheckpointRDD.scala +++ b/core/src/main/scala/spark/rdd/CheckpointRDD.scala @@ -80,12 +80,12 @@ private[spark] object CheckpointRDD extends Logging { val serializer = SparkEnv.get.serializer.newInstance() val serializeStream = serializer.serializeStream(fileOutputStream) serializeStream.writeAll(iterator) - fileOutputStream.close() + serializeStream.close() if (!fs.rename(tempOutputPath, finalOutputPath)) { if (!fs.delete(finalOutputPath, true)) { throw new IOException("Checkpoint failed: failed to delete earlier output of task " - + context.attemptId); + + context.attemptId) } if (!fs.rename(tempOutputPath, finalOutputPath)) { throw new IOException("Checkpoint failed: failed to save output of task: " @@ -119,7 +119,7 @@ private[spark] object CheckpointRDD extends Logging { val rdd = sc.makeRDD(1 to 10, 10).flatMap(x => 1 to 10000) val path = new Path(hdfsPath, "temp") val fs = path.getFileSystem(new Configuration()) - sc.runJob(rdd, CheckpointRDD.writeToFile(path.toString, 10) _) + sc.runJob(rdd, CheckpointRDD.writeToFile(path.toString, 1024) _) val cpRDD = new CheckpointRDD[Int](sc, path.toString) assert(cpRDD.splits.length == rdd.splits.length, "Number of splits is not the same") assert(cpRDD.collect.toList == rdd.collect.toList, "Data of splits not the same") |