如何在Spark Structured Streaming中用Trigger.Once()运行多流查询

注：Spark 3.3+ 版本推荐使用语义一致、性能更优的 Trigger.AvailableNow() 替代 Trigger.Once()，原有 Trigger.Once 已被官方标记为废弃。

核心实现步骤 #

• 独立配置每个流查询的存储隔离
  每个流查询必须分配独立的 checkpointLocation 路径，禁止多个查询共用同一路径，避免offset、状态元数据互相覆盖导致的任务报错、数据错乱。配置示例：

  import org.apache.spark.sql.streaming.Trigger
  
  // 流查询1配置
  val query1 = sourceDf1.writeStream
    .option("checkpointLocation", "/path/to/checkpoint/query1")
    .trigger(Trigger.Once())
    .format("parquet")
    .start("/path/to/sink/query1")
  
  // 流查询2配置
  val query2 = sourceDf2.writeStream
    .option("checkpointLocation", "/path/to/checkpoint/query2")
    .trigger(Trigger.Once())
    .format("kafka")
    .start()
  

• 按依赖关系控制查询启动顺序
  若多个流查询存在上下游依赖（如查询2的输入是查询1的输出），必须串行启动，前一个查询运行结束后再启动下一个，避免读取到不完整的中间数据。无依赖的查询可并行启动，需提前评估集群资源容量，避免资源争抢导致OOM或任务超时。串行启动示例：

  // 先运行依赖前置的查询1
  query1.awaitTermination()
  // 确认查询1运行成功后再启动查询2
  query2.awaitTermination()
  

• 单查询级别的异常捕获与重试
  给每个流查询单独包裹异常捕获逻辑，单个查询故障不影响其他查询的运行，也不会直接导致整个应用退出。故障重试前需校验对应checkpoint路径的元数据完整性，若为资源不足、网络波动等临时问题可直接重试，若为数据格式错误、Schema不兼容等问题需先修复问题再重试。
• 运行参数调优规避过载
  配置spark.sql.streaming.maxBatchesPerTrigger参数限制单次Trigger运行的最大批次量，避免一次性拉取过多数据导致内存溢出；配置spark.streaming.stopGracefullyOnShutdown=true保证任务停止时会完整提交当前批次的offset和状态，避免元数据损坏。
• 运行结果一致性校验
  所有查询运行结束后，通过每个查询的lastProgress属性获取本次运行的输入行数、offset提交范围等指标，和上下游数据量做校验，确认没有漏处理、重复处理的情况。

常见故障规避规则 #

• 禁止修改运行中查询的checkpoint路径和核心处理逻辑，如需调整要先测试再上线
• 数据源Schema发生变更时，提前配置mergeSchema=true或者指定读取Schema，避免Schema匹配错误导致任务崩溃
• 避免强制kill正在运行的流任务，尽量通过Spark的REST接口或者spark.stop()触发优雅关闭，防止元数据写一半损坏

内容的提问来源于stack exchange，提问作者Naga