You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by td...@apache.org on 2014/11/10 21:33:56 UTC
spark git commit: Update RecoverableNetworkWordCount.scala
Repository: spark
Updated Branches:
refs/heads/master 3a02d416c -> 0340c56a9
Update RecoverableNetworkWordCount.scala
Trying this example, I missed the moment when the checkpoint was iniciated
Author: comcmipi <pi...@fns.uniba.sk>
Closes #2735 from comcmipi/patch-1 and squashes the following commits:
b6d8001 [comcmipi] Update RecoverableNetworkWordCount.scala
96fe274 [comcmipi] Update RecoverableNetworkWordCount.scala
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/0340c56a
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/0340c56a
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/0340c56a
Branch: refs/heads/master
Commit: 0340c56a921d4eb4bc9058e25e926721f8df594c
Parents: 3a02d41
Author: comcmipi <pi...@fns.uniba.sk>
Authored: Mon Nov 10 12:33:48 2014 -0800
Committer: Tathagata Das <ta...@gmail.com>
Committed: Mon Nov 10 12:33:48 2014 -0800
----------------------------------------------------------------------
.../spark/examples/streaming/RecoverableNetworkWordCount.scala | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/0340c56a/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala
----------------------------------------------------------------------
diff --git a/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala b/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala
index eb48db8..19427e6 100644
--- a/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala
+++ b/examples/src/main/scala/org/apache/spark/examples/streaming/RecoverableNetworkWordCount.scala
@@ -56,7 +56,7 @@ import org.apache.spark.util.IntParam
*/
object RecoverableNetworkWordCount {
- def createContext(ip: String, port: Int, outputPath: String) = {
+ def createContext(ip: String, port: Int, outputPath: String, checkpointDirectory: String) = {
// If you do not see this printed, that means the StreamingContext has been loaded
// from the new checkpoint
@@ -66,6 +66,7 @@ object RecoverableNetworkWordCount {
val sparkConf = new SparkConf().setAppName("RecoverableNetworkWordCount")
// Create the context with a 1 second batch size
val ssc = new StreamingContext(sparkConf, Seconds(1))
+ ssc.checkpoint(checkpointDirectory)
// Create a socket stream on target ip:port and count the
// words in input stream of \n delimited text (eg. generated by 'nc')
@@ -101,7 +102,7 @@ object RecoverableNetworkWordCount {
val Array(ip, IntParam(port), checkpointDirectory, outputPath) = args
val ssc = StreamingContext.getOrCreate(checkpointDirectory,
() => {
- createContext(ip, port, outputPath)
+ createContext(ip, port, outputPath, checkpointDirectory)
})
ssc.start()
ssc.awaitTermination()
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org