Fix example

Author: tdas

examples/src/main/java/org/apache/spark/examples/streaming/JavaStatefulNetworkWordCount.java

@@ -90,19 +90,19 @@ public Tuple2<String, Integer> call(String s) {
         });
 
     // Update the cumulative count function
-    final Function3<String, Optional<Integer>, State<Integer>, Optional<Tuple2<String, Integer>>> mappingFunc =
-        new Function3<String, Optional<Integer>, State<Integer>, Optional<Tuple2<String, Integer>>>() {
+    final Function3<String, Optional<Integer>, State<Integer>, Tuple2<String, Integer>> mappingFunc =
+        new Function3<String, Optional<Integer>, State<Integer>, Tuple2<String, Integer>>() {
 
           @Override
-          public Optional<Tuple2<String, Integer>> call(String word, Optional<Integer> one, State<Integer> state) {
+          public Tuple2<String, Integer> call(String word, Optional<Integer> one, State<Integer> state) {
             int sum = one.or(0) + (state.exists() ? state.get() : 0);
             Tuple2<String, Integer> output = new Tuple2<String, Integer>(word, sum);
             state.update(sum);
-            return Optional.of(output);
+            return output;
           }
         };
 
-    // This will give a Dstream made of state (which is the cumulative count of the words)
+    // DStream made of get cumulative counts that get updated in every batch
     JavaMapWithStateDStream<String, Integer, Integer, Tuple2<String, Integer>> stateDstream =
         wordsDstream.mapWithState(StateSpec.function(mappingFunc).initialState(initialRDD));
 

streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala

@@ -61,8 +61,7 @@ import org.apache.spark.{HashPartitioner, Partitioner}
  *       };
  *
  *    JavaMapWithStateDStream<String, Integer, Integer, String> mapWithStateDStream =
- *        keyValueDStream.<Integer, String>mapWithState(
- *            StateSpec.function(mappingFunction).numPartitions(10));
+ *        keyValueDStream.mapWithState(StateSpec.function(mappingFunc));
  * }}}
  *
  * @tparam KeyType    Class of the state key
@@ -137,9 +136,8 @@ sealed abstract class StateSpec[KeyType, ValueType, StateType, MappedType] exten
  *       };
  *
  *    JavaMapWithStateDStream<String, Integer, Integer, String> mapWithStateDStream =
- *        keyValueDStream.<Integer, String>mapWithState(
- *            StateSpec.function(mappingFunction).numPartitions(10));
- * }}}
+ *        keyValueDStream.mapWithState(StateSpec.function(mappingFunc));
+ *}}}
  */
 @Experimental
 object StateSpec {

streaming/src/main/scala/org/apache/spark/streaming/api/java/JavaPairDStream.scala

@@ -449,9 +449,8 @@ class JavaPairDStream[K, V](val dstream: DStream[(K, V)])(
    *       };
    *
    *    JavaMapWithStateDStream<String, Integer, Integer, String> mapWithStateDStream =
-   *        keyValueDStream.<Integer, String>mapWithState(
-   *            StateSpec.function(mappingFunction).numPartitions(10));
-   * }}}
+   *        keyValueDStream.mapWithState(StateSpec.function(mappingFunc));
+   *}}}
    *
    * @param spec          Specification of this transformation
    * @tparam StateType    Class type of the state data