scala - ERROR Uncaught throwable from user code : java. lang.IllegalStateException in spark Streaming

Question

我有 2 个流式查询，如下所示

val streamingQuery = injectableDependencies.writeStreamDfToDeltaFile(validDF, dataPath, checkPointPath, configuration.pollingTimerSeconds, queryName, configuration.outputMode, configuration.partitionBy)

//streamingQuery for Invalid tables
val streamingQueryInvalid = injectableDependencies.writeStreamDfToDeltaFile(inValidDF, dataPathInvalid, checkPointPath, configuration.pollingTimerSeconds, queryNameInvalid, configuration.outputMode, configuration.partitionBy)

//creating valid tables
injectableDependencies.createTableOverDeltaFile(configuration.tableName, configuration.databaseName, dataPath, streamingQuery)

//creating invalid tables
injectableDependencies.createTableOverDeltaFile("Invalid"+configuration.tableName, configuration.databaseName, dataPath, streamingQueryInvalid)

下面我正在尝试写流

  df
  .writeStream
  .format("DELTA")
  .option("path", dataPath)
  .option("checkpointLocation", checkPointPath)
  .partitionBy(partitionBy.getOrElse(List[String]()): _*)
  .outputMode(outputMode)
  .trigger(Trigger.ProcessingTime(pollingTimerSeconds.seconds))
  .queryName(queryName)
  .start()

出现以下错误

ERROR Uncaught throwable from user code: java.lang.IllegalStateException: Cannot start query with id 74e6b948-bc55-419c-af42-34ef7ea015ba as another query with same id is already active. Perhaps you are attempting to restart a query from the checkpoint that is already active.

谁能告诉我哪里出了问题

Answer 1

我遇到了同样的问题，并通过确保每个流式查询都将其专用的 checkpointLocation 作为 writeStream 调用中的选项来解决。