改进代码：import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream} object UpdateStateByKeyTest { //newValues表示当前批次汇总成的（K,V）中相同K的所有V //runningCount表示历史的所有相同key的value总和 def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = { val newCount = runningCount.getOrElse(0) + newValues.sum Some(newCount) } def main(args: Array[String]): Unit = { //1.创建SparkConf对象 val sparkConf: SparkConf = new SparkConf().setAppName("UpdateStateByKeyTest").setMaster("local[2]") //2.创建SparkContext对象 val sc: SparkContext = new SparkContext(sparkConf) //3.设置日志级别 sc.setLogLevel("WARN") //4.创建StreamingContext，两个参数：1.SparkContext对象 2.批处理时间间隔 val ssc: StreamingContext = new StreamingContext(sc, Seconds(5)) //5.配置检查点目录，使用updateStateByKey()方法必须配置检查点目录 ssc.checkpoint("./") //6.连接socket服务，需要socket的地址，端口号，存储级别 val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.92.131", 9999) //7.按空格切分每一行，并且将切分出来的单词出现的次数记录为1 val wordAndOne: DStream[(String, Int)] = dstream.flatMap(_.split(" ")).map(word => (word, 1)) //8.调用UpdateStateByKey操作，统计每个单词在全局中出现的次数 val result: DStream[(String,Int)] = wordAndOne.updateStateByKey(updateFunction) //9.打印输出结果 result.print() //10.开启流式计算 ssc.start() //11.用于保持程序一直运行，除非人为干预停止 ssc.awaitTermination() } }

改进代码如下：

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object UpdateStateByKeyTest {
  //newValues表示当前批次汇总成的（K,V）中相同K的所有V
  //runningCount表示历史的所有相同key的value总和
  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val newCount = runningCount.getOrElse(0) + newValues.sum
    Some(newCount)
  }

  def main(args: Array[String]): Unit = {
    //1.创建StreamingContext，两个参数：1.SparkConf对象 2.批处理时间间隔
    val ssc: StreamingContext = new StreamingContext(new SparkConf().setAppName("UpdateStateByKeyTest").setMaster("local[2]"), Seconds(5))

    //2.设置日志级别
    ssc.sparkContext.setLogLevel("WARN")

    //3.配置检查点目录，使用updateStateByKey()方法必须配置检查点目录
    ssc.checkpoint("./")

    //4.连接socket服务，需要socket的地址，端口号，存储级别
    val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.92.131", 9999)

    //5.按空格切分每一行，并且将切分出来的单词出现的次数记录为1
    val wordAndOne: DStream[(String, Int)] = dstream.flatMap(_.split(" ")).map(word => (word, 1))

    //6.调用UpdateStateByKey操作，统计每个单词在全局中出现的次数
    val result: DStream[(String,Int)] = wordAndOne.updateStateByKey(updateFunction)

    //7.打印输出结果
    result.print()

    //8.开启流式计算
    ssc.start()

    //9.用于保持程序一直运行，除非人为干预停止
    ssc.awaitTermination()
  }
}

改进的地方如下：

1. 将SparkConf和SparkContext合并为一个语句，简化代码。
2. 将StreamingContext作为主要的上下文对象，避免在更新状态操作(updateStateByKey)时出现问题。
3. 调整了代码结构，将一些常规设置（如日志级别和检查点目录）放在主要逻辑之前。