Kate Eri created SPARK-20325:
--------------------------------
Summary: Spark Structured Streaming documentation Update:
checkpoint configuration
Key: SPARK-20325
URL: https://issues.apache.org/jira/browse/SPARK-20325
Project: Spark
Issue Type: Bug
Components: Structured Streaming
Affects Versions: 2.1.0
Reporter: Kate Eri
Priority: Minor
I have configured the following stream outputting to Kafka:
{code}
map.foreach(metric => {
streamToProcess
.groupBy(metric)
.agg(count(metric))
.writeStream
.outputMode("complete")
.option("checkpointLocation", checkpointDir)
.foreach(kafkaWriter)
.start()
})
{code}
And configured the checkpoint Dir for each of output sinks like:
.option("checkpointLocation", checkpointDir) according to the documentation =>
http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#recovering-from-failures-with-checkpointing
As a result I've got the following exception:
Cannot start query with id bf6a1003-6252-4c62-8249-c6a189701255 as another
query with same id is already active. Perhaps you are attempting to restart a
query from checkpoint that is already active.
java.lang.IllegalStateException: Cannot start query with id
bf6a1003-6252-4c62-8249-c6a189701255 as another query with same id is already
active. Perhaps you are attempting to restart a query from checkpoint that is
already active.
at
org.apache.spark.sql.streaming.StreamingQueryManager.startQuery(StreamingQueryManager.scala:291)
So according to current spark logic for “foreach” sink the checkpoint
configuration is loaded in the following way:
{code:title=StreamingQueryManager.scala}
val checkpointLocation = userSpecifiedCheckpointLocation.map { userSpecified
=>
new Path(userSpecified).toUri.toString
}.orElse {
df.sparkSession.sessionState.conf.checkpointLocation.map { location =>
new Path(location,
userSpecifiedName.getOrElse(UUID.randomUUID().toString)).toUri.toString
}
}.getOrElse {
if (useTempCheckpointLocation) {
Utils.createTempDir(namePrefix = s"temporary").getCanonicalPath
} else {
throw new AnalysisException(
"checkpointLocation must be specified either " +
"""through option("checkpointLocation", ...) or """ +
s"""SparkSession.conf.set("${SQLConf.CHECKPOINT_LOCATION.key}",
...)""")
}
}
{code}
so first spark take checkpointDir from query, then from sparksession
(spark.sql.streaming.checkpointLocation) and so on.
But this behavior was not documented, thus two questions:
1) could we update documentation for Structured Streaming and describe this
behavior
2) Do we really need to specify the checkpoint dir per query? what the reason
for this? finally we will be forced to write some checkpointDir name generator,
for example associate it with some particular named query and so on?
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
