使用scala编写 Spark-Stream 滑动窗口3小时. 滑动步长五分钟.读取Hive表数据 写一个统计该窗口内所有用户的订单金额

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

// 创建SparkSession
val spark = SparkSession.builder()
  .appName("Windowed Order Amount")
  .enableHiveSupport()
  .getOrCreate()

// 设定滑动窗口大小和步长
val windowSize = "3 hours"
val slideInterval = "5 minutes"

// 读取Hive表数据
val orders = spark.sql("SELECT * FROM orders")

// 定义窗口
val windowedOrders = orders
  .withWatermark("order_date", "1 hour") // 设定水位线为1小时，保证数据的时效性
  .groupBy(
    window($"order_date", windowSize, slideInterval),
    $"user_id"
  )
  .agg(sum($"order_amount").as("total_order_amount"))
  .select($"window.start".as("start_time"), $"window.end".as("end_time"), $"user_id", $"total_order_amount")

// 输出结果
val query = windowedOrders.writeStream
  .outputMode("complete")
  .format("console")
  .start()

query.awaitTermination()