Problems with the state functions in Spark Streaming
<p>I try to use Spark Streaming and want to have a global state object that can be updated after each batch was processed. As far as I have found, there are at least two options for me:
1. Use <code>mapWithState</code>, where Spark will update the state automatically after each batch was processed
2. use the <code>updateStateByKey</code> function, where I have to call the update myself</p>

<p>For me both cases would be fine, but I get the same error both of the times. Here is my example code for both cases and their error:</p>

<pre><code>    Function3&lt;String, Optional&lt;Integer&gt;, State&lt;Integer&gt;, Tuple2&lt;String, Integer&gt;&gt; mappingFunc =
    new Function3&lt;String, Optional&lt;Integer&gt;, State&lt;Integer&gt;, Tuple2&lt;String, Integer&gt;&gt;() {
      @Override
      public Tuple2&lt;String, Integer&gt; call(String word, Optional&lt;Integer&gt; one,
          State&lt;Integer&gt; state) {
        int sum = one.orElse(0) + (state.exists() ? state.get() : 0);
        Tuple2&lt;String, Integer&gt; output = new Tuple2&lt;&gt;(word, sum);
        state.update(sum);
        return output;
      }
    };


    JavaMapWithStateDStream&lt;String, Integer, Integer, Tuple2&lt;String, Integer&gt;&gt; stateDstream =
    wordsDstream.mapWithState(StateSpec.function(mappingFunc).initialState(initialRDD));;
               Tuple2&lt;String, Long&gt; output = new Tuple2&lt;&gt;(word, sum);
               state.update(sum);
               return new String("Test");
        }
    });  
</code></pre>

<p>This one is taken from the example provodided by Spark itself. Sadly I recieve the following error regarding the<code>StateSpec.function(...)</code>: </p>

<blockquote>
  <p>The method function(Function3&lt; KeyType,Optional&lt; ValueType>,State&lt; StateType>,MappedType>) in the type StateSpec is not applicable for the arguments (Function3&lt; String,Optional&lt; Integer>,State&lt; Integer>,Tuple2&lt; String,Integer>>)</p>
</blockquote>

<p>Using:</p>

<pre><code>JavaPairDStream&lt;String, Long&gt; runningCounts = processed.updateStateByKey(new Function2&lt;List&lt;Long&gt;, Optional&lt;Long&gt;, Optional&lt;Long&gt;&gt;() {
            public Optional&lt;Long&gt; call(List&lt;Long&gt; values, Optional&lt;Long&gt; state) {
                Long newSum = state.orElse((long)0);
                for(long x : values){
                    newSum +=x;
                }
                return Optional.of(newSum);
              });
</code></pre>

<p>Results in a similar error:</p>

<blockquote>
  <p>The method updateStateByKey(Function2&lt; List&lt; Long>,Optional&lt; S>,Optional&lt; S>>) in the type JavaPairDStream&lt; String,Long> is not applicable for the arguments (new Function2&lt; List,Optional&lt; Long>,Optional&lt; Long>>(){})</p>
</blockquote>

<p>A Snapshot of my imports are:</p>

<pre><code>import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.Function3;
import org.apache.spark.api.java.function.Function4;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.api.java.function.VoidFunction2;
</code></pre>

<p>I hope someone can help me to find my mistake.</p>