Update checkpoint call as in apache#2735

markhamstra · Nov 7, 2014 · 0d0bf29 · 0d0bf29
1 parent 35f23e3
commit 0d0bf29
Showing 1 changed file with 7 additions and 3 deletions.
diff --git a/...es/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java b/...es/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java
@@ -69,7 +69,10 @@
 public final class JavaRecoverableNetworkWordCount {
   private static final Pattern SPACE = Pattern.compile(" ");
 
-  private static JavaStreamingContext createContext(String ip, int port, String outputPath) {
+  private static JavaStreamingContext createContext(String ip,
+                                                    int port,
+                                                    String checkpointDirectory,
+                                                    String outputPath) {
 
     // If you do not see this printed, that means the StreamingContext has been loaded
     // from the new checkpoint
@@ -81,6 +84,7 @@ private static JavaStreamingContext createContext(String ip, int port, String ou
     SparkConf sparkConf = new SparkConf().setAppName("JavaRecoverableNetworkWordCount");
     // Create the context with a 1 second batch size
     JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.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')
@@ -135,12 +139,12 @@ public static void main(String[] args) {
 
     final String ip = args[0];
     final int port = Integer.parseInt(args[1]);
-    String checkpointDirectory = args[2];
+    final String checkpointDirectory = args[2];
     final String outputPath = args[3];
     JavaStreamingContextFactory factory = new JavaStreamingContextFactory() {
       @Override
       public JavaStreamingContext create() {
-        return createContext(ip, port, outputPath);
+        return createContext(ip, port, checkpointDirectory, outputPath);
       }
     };
     JavaStreamingContext ssc = JavaStreamingContext.getOrCreate(checkpointDirectory, factory);