apache-spark - 如何在单个 Spark 应用程序中启动多个流式查询？

Question

我已经构建了几个 Spark Structured Streaming 查询以在 EMR 上运行，它们是长时间运行的查询，并且需要始终运行，因为它们都是 ETL 类型的查询，当我在 EMR 上向 YARN 集群提交作业时，我可以提交一个单 Spark 应用。因此， Spark 应用程序应该有多个流式查询。

我对如何以编程方式在同一提交中构建/启动多个流查询感到困惑。

例如:我有这个代码:

case class SparkJobs(prop: Properties) extends Serializable {
  def run() = {
      Type1SparkJobBuilder(prop).build().awaitTermination()
      Type1SparkJobBuilder(prop).build().awaitTermination()
  }
}

我用 SparkJobs(new Properties()).run() 在我的主类开火

当我在 spark 历史服务器中看到时，只有第一个 spark 流作业 (Type1SparkJob) 正在运行。

以编程方式在同一个 spark 提交中触发多个流查询的推荐方法是什么，我也找不到合适的文档。

Answer 1

既然你调用 awaitTermination在第一个查询中，它将阻塞，直到它在开始第二个查询之前完成。所以你想启动两个查询，然后使用 StreamingQueryManager.awaitAnyTermination .

val query1 = df.writeStream.start()
val query2 = df.writeStream.start()

spark.streams.awaitAnyTermination()

除了上述之外，默认情况下 Spark 使用 FIFO 调度程序。这意味着第一个查询在执行时获取集群中的所有资源。由于您尝试同时运行多个查询，您应该切换到 FAIR scheduler

如果您有一些查询应该比其他查询拥有更多的资源，那么您还可以调整各个调度程序池。