上面的示例是dstream示例，而不是结构化流。您需要将spark结构化流看作是将数据加载到一个无界表中。
假设数据源是kafka，下面是结构化流的一个基本示例。请注意，使用readstream和writestream api无法进行模式推断。模式需要来自数据源连接器，在本例中是kafka。

val df = sparkSession
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9031")
  .option("subscribe", "word-count")
  .option("startingOffsets", "latest")
  .option("failOnDataLoss", "false")
  .load()

val query = df
  .selectExpr("CAST(key AS STRING) as key", "CAST(value AS STRING) as value")
  .as[(String, String)]
  .select(from_json($"value", schema = schema).as("data"))
  .writeStream
  .format("parquet")
  .option("path", "/parquet/word-count/")
  .option("checkpointLocation", "/tmp/word-count-chkpnt")
  .trigger(ProcessingTime("10 second"))
  .outputMode(OutputMode.Append())
  .start()

使用.trigger（）函数创建微批，outputmode保存每个微批的结果。在这个例子中，我每10秒创建一个微批， .trigger(ProcessingTime("10 second")) 以及将流中的每个事件作为一行附加到parquet文件 .outputMode(OutputMode.Append()) 在您的例子中，您需要使用.trigger（），并选择一个微批处理间隔 .outputMode(outputMode.Update()) 插入具有值的新键或使用递增值更新现有键。
下面的部分是聚合逻辑的发展方向。您可以将聚合逻辑分解为单独的Dataframe，并将Dataframe作为流写入，而不是为了可读性而进行链接。

.selectExpr("CAST(key AS STRING) as key", "CAST(value AS STRING) as value")
  .as[(String, String)]
  .select(from_json($"value", schema = schema).as("data"))

结构化流媒体的另一个例子。