Structured Streaming filter only latest record per key

latest.scala

spark
      .readStream
      .format("socket")
      .option("host", "127.0.0.1")
      .option("port", 9999)
      .load()
      .as[String]
      .map { value =>
        val split = value.split(",")
        (split(0), split(1))
      }
      .groupByKey(_._1)
      .mapGroupsWithState { (_, tuples: Iterator[(String, String)], state: GroupState[(String, String)]) =>
        val list = tuples.toList
        val max = state.getOption.fold(list)(_ :: list).maxBy(_._2.toInt)
        state.update(max)
        state.get
      }
      .writeStream
      .outputMode("update")
      .trigger(Trigger.ProcessingTime(3000))
      .format("console")
      .start()