Azure Databricks Spark Structured Streaming任务运行5分钟后莫名终止

问题描述 #

在Azure Databricks上运行简化版Spark Structured Streaming任务：从Kafka主题读取数据，通过forEachBatch输出到控制台。本地环境可无限正常运行，但在Databricks上，任务刚运行满5分钟就会以Cancelled状态终止，日志无报错信息，仅出现以下优雅关闭相关日志：

22/11/04 18:31:30 INFO DriverCorral$: Cleaning the wrapper ReplId-1ea30-8e4c0-48422-a (currently in status Running(ReplId-1ea30-8e4c0-48422-a,ExecutionId(job-774316032912321-run-84401-action-5645198327600153),RunnableCommandId(9102993760433650959)))
22/11/04 18:31:30 INFO DAGScheduler: Asked to cancel job group 2207618020913201706_9102993760433650959_job-774316032912321-run-84401-action-5645198327600153
22/11/04 18:31:30 INFO ScalaDriverLocal: cancelled jobGroup:2207618020913201706_9102993760433650959_job-774316032912321-run-84401-action-5645198327600153 
22/11/04 18:31:30 INFO ScalaDriverWrapper: Stopping streams for commandId pattern: CommandIdPattern(2207618020913201706,None,Some(job-774316032912321-run-84401-action-5645198327600153)).
22/11/04 18:31:30 INFO DatabricksStreamingQueryListener: Stopping the stream [id=d41eff2a-4de6-4f17-8d1c-659d1c1b8d98, runId=5bae9fb4-b5e1-45a0-af1e-a2f2553592c9]
22/11/04 18:31:30 INFO DAGScheduler: Asked to cancel job group 5bae9fb4-b5e1-45a0-af1e-a2f2553592c9
22/11/04 18:31:30 INFO TaskSchedulerImpl: Cancelling stage 366
22/11/04 18:31:30 INFO TaskSchedulerImpl: Killing all running tasks in stage 366: Stage cancelled
22/11/04 18:31:30 INFO MicroBatchExecution: QueryExecutionThread.interruptAndAwaitExecutionThreadTermination called with streaming query exit timeout=15000 ms

参考代码：

val incomingStream = spark.readStream
  .format("kafka")
  .option("subscribe",ehName)
  .option("kafka.bootstrap.servers",topicUriWithPort)
  .option("kafka.sasl.mechanism","PLAIN")
  .option("kafka.security.protocol","SASL_SSL")
  .option("kafka.sasl.jaas.config",jaas)
  .option("startingOffsets", "earliest") 
  .option("failOnDataLoss","false")
  .option("maxOffsetsPerTrigger", 1) //todo make config
  .load()

val processedWriteStream = incomingStream
  .writeStream
  .queryName("query2")
  .foreachBatch((d: DataFrame, b: Long) => {
    d.show()
  })
  .start()
processedWriteStream.awaitTermination()

原因及解决方案 #

核心原因 #

Databricks笔记本默认对单个命令的执行时间有5分钟限制，超时后系统会自动终止命令对应的任务，日志中的DriverCorral$: Cleaning the wrapper就是系统清理超时命令的标志。

解决方法 #

1. 调整笔记本命令超时时间
   进入Databricks管理控制台 -> 工作区设置，找到“笔记本命令超时”选项，根据需求延长超时时长（如设置为几小时）。注意此设置为全局生效，会影响所有笔记本命令。

2. 以Databricks作业模式运行流式任务
   流式任务属于长运行任务，更适合通过作业而非笔记本命令执行：

   • 将代码打包为Jar，或直接在Databricks作业中创建任务，指定集群运行。
   • 作业模式无单命令超时限制，支持任务长期运行。

3. 优化任务触发配置
   代码中maxOffsetsPerTrigger=1会导致任务触发过于频繁，可适当调大该值，减少触发次数，同时保证任务在超时时间内有持续执行动作。

4. 检查集群自动终止设置
   确认集群的“自动终止”时间是否过短，若集群在5分钟内无活跃任务会自动终止，也会导致流式任务被取消。可延长自动终止时间，或在测试环境关闭该功能。

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