val dstream = KafkaUtils.createDirectStream[String, String](
        ssc,
        PreferConsistent,
        Subscribe[String, String](Array(top), kafkaParams)
    ).map(_.value()).window(Seconds(5),Seconds(5))
    //按照进来的每一条流进行实时统计（5s）一次

   dstream.map(row => (row,1L)).reduceByKey(_+_).updateStateByKey[Int](updateFunction _)
    ssc.start()
    ssc.awaitTermination()
/**
  * 当前数据去更新已有的或者老的数据
  * @param currentValues 当前的
  * @param preValues 后面的
  */
//定义了一个求和的函数
def updateFunction(currentValues: Seq[Int], preValues: Option[Int]): Unit ={
    val current = currentValues.sum
    //因为累计加值所以有可能为空，如果为空的话则返回为none，所以给返回值为0
    val pre = preValues.getOrElse(0)
    Some(current + pre)