From 7c304f671d2f106c7ce5bc940540ff1a346df8f7 Mon Sep 17 00:00:00 2001 From: Gabor Somogyi Date: Mon, 4 Feb 2019 12:42:07 +0100 Subject: [PATCH 1/3] [SPARK-26389][SS] Add force delete temp checkpoint configuration --- .../apache/spark/sql/internal/SQLConf.scala | 7 ++++ .../execution/streaming/StreamExecution.scala | 11 +++++-- .../sql/streaming/StreamingQueryManager.scala | 8 +++-- .../test/DataStreamReaderWriterSuite.scala | 32 ++++++++++++++++++- 4 files changed, 52 insertions(+), 6 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index 11e1a5eefe933..f378986d5e8d0 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -907,6 +907,13 @@ object SQLConf { .stringConf .createOptional + val FORCE_DELETE_TEMP_CHECKPOINT_LOCATION = + buildConf("spark.sql.streaming.forceDeleteTempCheckpointLocation") + .doc("Temporary checkpoint locations deleted normally when the query didn't fail." + + " When true, it will be deleted even if the query failed.") + .booleanConf + .createWithDefault(false) + val MIN_BATCHES_TO_RETAIN = buildConf("spark.sql.streaming.minBatchesToRetain") .internal() .doc("The minimum number of batches that must be retained and made recoverable.") diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala index 90f7b477103ae..dc5411780ac4f 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala @@ -55,7 +55,8 @@ case object RECONFIGURING extends State * and the results are committed transactionally to the given [[Sink]]. * * @param deleteCheckpointOnStop whether to delete the checkpoint if the query is stopped without - * errors + * errors. Checkpoint deletion can be forced with the appropriate + * Spark configuration. */ abstract class StreamExecution( override val sparkSession: SparkSession, @@ -92,6 +93,7 @@ abstract class StreamExecution( fs.mkdirs(checkpointPath) checkpointPath.makeQualified(fs.getUri, fs.getWorkingDirectory).toUri.toString } + logInfo(s"Checkpoint root $checkpointRoot resolved to $resolvedCheckpointRoot.") def logicalPlan: LogicalPlan @@ -335,10 +337,13 @@ abstract class StreamExecution( postEvent( new QueryTerminatedEvent(id, runId, exception.map(_.cause).map(Utils.exceptionString))) - // Delete the temp checkpoint only when the query didn't fail - if (deleteCheckpointOnStop && exception.isEmpty) { + // Delete the temp checkpoint when either force delete enabled or the query didn't fail + if (deleteCheckpointOnStop && + (sparkSession.sessionState.conf.getConf(SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION) || + exception.isEmpty)) { val checkpointPath = new Path(resolvedCheckpointRoot) try { + logInfo(s"Deleting checkpoint $checkpointPath.") val fs = checkpointPath.getFileSystem(sparkSession.sessionState.newHadoopConf()) fs.delete(checkpointPath, true) } catch { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala index 881cd96cc9dc9..054d3d468e8ef 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala @@ -221,9 +221,13 @@ class StreamingQueryManager private[sql] (sparkSession: SparkSession) extends Lo } }.getOrElse { if (useTempCheckpointLocation) { - // Delete the temp checkpoint when a query is being stopped without errors. deleteCheckpointOnStop = true - Utils.createTempDir(namePrefix = s"temporary").getCanonicalPath + val tempDir = Utils.createTempDir(namePrefix = s"temporary").getCanonicalPath + logWarning("Temporary checkpoint location created which is deleted normally when" + + s" the query didn't fail: $tempDir. If it's required to delete it under any" + + s" circumstances please set ${SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION.key} to" + + s" true. Important to know deleting temp checkpoint folder is best effort.") + tempDir } else { throw new AnalysisException( "checkpointLocation must be specified either " + diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamReaderWriterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamReaderWriterSuite.scala index 74ea0bfacba54..c3c7dcbaaece7 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamReaderWriterSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/test/DataStreamReaderWriterSuite.scala @@ -614,6 +614,21 @@ class DataStreamReaderWriterSuite extends StreamTest with BeforeAndAfter { } } + test("configured checkpoint dir should not be deleted if a query is stopped without errors and" + + " force temp checkpoint deletion enabled") { + import testImplicits._ + withTempDir { checkpointPath => + withSQLConf(SQLConf.CHECKPOINT_LOCATION.key -> checkpointPath.getAbsolutePath, + SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION.key -> "true") { + val ds = MemoryStream[Int].toDS + val query = ds.writeStream.format("console").start() + assert(checkpointPath.exists()) + query.stop() + assert(checkpointPath.exists()) + } + } + } + test("temp checkpoint dir should be deleted if a query is stopped without errors") { import testImplicits._ val query = MemoryStream[Int].toDS.writeStream.format("console").start() @@ -627,6 +642,17 @@ class DataStreamReaderWriterSuite extends StreamTest with BeforeAndAfter { } testQuietly("temp checkpoint dir should not be deleted if a query is stopped with an error") { + testTempCheckpointWithFailedQuery(false) + } + + testQuietly("temp checkpoint should be deleted if a query is stopped with an error and force" + + " temp checkpoint deletion enabled") { + withSQLConf(SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION.key -> "true") { + testTempCheckpointWithFailedQuery(true) + } + } + + private def testTempCheckpointWithFailedQuery(checkpointMustBeDeleted: Boolean): Unit = { import testImplicits._ val input = MemoryStream[Int] val query = input.toDS.map(_ / 0).writeStream.format("console").start() @@ -638,7 +664,11 @@ class DataStreamReaderWriterSuite extends StreamTest with BeforeAndAfter { intercept[StreamingQueryException] { query.awaitTermination() } - assert(fs.exists(checkpointDir)) + if (!checkpointMustBeDeleted) { + assert(fs.exists(checkpointDir)) + } else { + assert(!fs.exists(checkpointDir)) + } } test("SPARK-20431: Specify a schema by using a DDL-formatted string") { From ca77ebce1e807026cf2475f437c9a2ac31b6c652 Mon Sep 17 00:00:00 2001 From: Gabor Somogyi Date: Thu, 7 Feb 2019 15:37:50 +0100 Subject: [PATCH 2/3] Doc fic --- .../src/main/scala/org/apache/spark/sql/internal/SQLConf.scala | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala index f378986d5e8d0..d285e007dac1b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala @@ -909,8 +909,7 @@ object SQLConf { val FORCE_DELETE_TEMP_CHECKPOINT_LOCATION = buildConf("spark.sql.streaming.forceDeleteTempCheckpointLocation") - .doc("Temporary checkpoint locations deleted normally when the query didn't fail." + - " When true, it will be deleted even if the query failed.") + .doc("When true, enable temporary checkpoint locations force delete.") .booleanConf .createWithDefault(false) From ce2577d3e160f31272ae2cc34e2fc6fb56e6f218 Mon Sep 17 00:00:00 2001 From: Gabor Somogyi Date: Thu, 7 Feb 2019 21:30:31 +0100 Subject: [PATCH 3/3] Fix --- .../spark/sql/execution/streaming/StreamExecution.scala | 4 ++-- .../apache/spark/sql/streaming/StreamingQueryManager.scala | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala index dc5411780ac4f..dc9ed80b64205 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala @@ -339,8 +339,8 @@ abstract class StreamExecution( // Delete the temp checkpoint when either force delete enabled or the query didn't fail if (deleteCheckpointOnStop && - (sparkSession.sessionState.conf.getConf(SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION) || - exception.isEmpty)) { + (sparkSession.sessionState.conf + .getConf(SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION) || exception.isEmpty)) { val checkpointPath = new Path(resolvedCheckpointRoot) try { logInfo(s"Deleting checkpoint $checkpointPath.") diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala index 054d3d468e8ef..cb9ca4c59b8e4 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryManager.scala @@ -225,7 +225,7 @@ class StreamingQueryManager private[sql] (sparkSession: SparkSession) extends Lo val tempDir = Utils.createTempDir(namePrefix = s"temporary").getCanonicalPath logWarning("Temporary checkpoint location created which is deleted normally when" + s" the query didn't fail: $tempDir. If it's required to delete it under any" + - s" circumstances please set ${SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION.key} to" + + s" circumstances, please set ${SQLConf.FORCE_DELETE_TEMP_CHECKPOINT_LOCATION.key} to" + s" true. Important to know deleting temp checkpoint folder is best effort.") tempDir } else {