我试图用kafka实现wordcount，但在使用mapwithstate函数时出现了“类型不匹配”错误。
这是我的密码：

// make a connection to Kafka and read (key, value) pairs from it
    val sparkConf = new SparkConf().setAppName("DirectKafkaAvg").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(2))
    val kafkaConf = Map(
        "metadata.broker.list" -> "localhost:9092",
        "zookeeper.connect" -> "localhost:2181",
        "group.id" -> "kafka-spark-streaming",
        "zookeeper.connection.timeout.ms" -> "1000")
    val topics = Set("avg")
    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaConf, topics)
    val value = messages.map{case (key, value) => value.split(',')}
    val pairs = value.map(record => (record(1), record(2)))

    // measure the average value for each key in a stateful manner
    def mappingFunc(key: String, value: Option[Double], state: State[Double]): Option[(String, Double)] = {
        val sum = value.getOrElse(0.0) + state.getOption.getOrElse(0.0)
        val output = Option(key, sum)
        state.update(sum)
        output
    }
    val spec = StateSpec.function(mappingFunc _)
    val stateDstream = pairs.mapWithState(spec)

    // store the result in Cassandra
    stateDstream.print()

    ssc.start()
    ssc.awaitTermination()

错误日志如下：

[error] KafkaSpark.scala:50: type mismatch;
[error]  found   : org.apache.spark.streaming.StateSpec[String,Double,Double,Option[(String, Double)]]
[error]  required: org.apache.spark.streaming.StateSpec[String,String,?,?]
[error]     val stateDstream = pairs.mapWithState(spec)
[error]                                           ^
[error] one error found
[error] (compile:compileIncremental) Compilation failed

有人知道怎么处理吗？