使用Spark structured streaming仅保留最新数据

import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder
  .appName("StructuredStreamingExample")
  .master("local[*]")
  .getOrCreate()

val inputData = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "topic_name")
  .load()

import org.apache.spark.sql.functions._

val processedData = inputData
  .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .groupBy("key")
  .agg(max("value") as "latest_value")

val query = processedData
  .writeStream
  .outputMode("update")
  .format("console")
  .start()

query.awaitTermination()