Fixes #91 Delete temporary files between RDDs

phymbert · Sep 28, 2021 · 4197e8c · 4197e8c
1 parent d79ea0c
commit 4197e8c
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/core/src/main/scala/org/apache/spark/search/rdd/SearchRDDIndexer.scala b/core/src/main/scala/org/apache/spark/search/rdd/SearchRDDIndexer.scala
@@ -57,6 +57,7 @@ private[search] class SearchRDDIndexer[S: ClassTag](sc: SparkContext,
   protected def streamPartitionIndexZip(context: TaskContext, searchRDDPartition: SearchPartitionIndex[S]): Iterator[Array[Byte]] = {
     val localIndexDirPath = new File(searchRDDPartition.indexDir)
     val targetPath = new File(localIndexDirPath.getParent, s"${localIndexDirPath.getName}.zip")
+    context.addTaskCompletionListener[Unit](_ => FileUtils.delete(targetPath))
     zipPartition(localIndexDirPath.toPath, new FileOutputStream(targetPath))
 
     new InterruptibleIterator[Array[Byte]](context, new FileInputStreamIterator(targetPath))