cleanSource option does not delete any files

Viewed 641

I have a Structured Streaming Job with Trigger.Once() enabled which I run each 20 minutes. After each running, I wat remove my processed parquet files from S3, so I enabled the cleanSource delete option, but it does not work and I don't know why !

Before showing my code, I have to comment about him. I'm running multiple structured streaming queries in paralell, I have 5 buckets and I submit this in parallel. The job works perfectly, but does not delete any processed files.

var table = ['table1','table2','table3','table4','table5']
tables.par.map(table => {
        new ReplicationTables().run(table)
      })

object ReplicationTables {

def run(table): Unit = {
 val dataFrame = spark.readStream
      .option("mergeSchema", "true")
      .schema(dfSchema)
      .option("cleanSource","delete")  
      .parquet(s"s3a://my-bucket/${table}/*")

// I do some transformation and after I write my new dataframe called df to S3 in Delta format

df.writeStream
      .format("delta")
      .outputMode("append")
      .queryName(s"Delta/${table.schema}/${table.name}")
      .trigger(Trigger.Once())
     .option("checkpointLocation", s"s3a://my-bucket/checkpoints/${table.schema}/${table.name}")
      .start(s"s3a://my-bucket/Delta_Tables/${table}/")
      .awaitTermination()



}

}

PS: Even with INFO log level I does not have any logs about the cleanSource

PS 2: Follow the docs of Structured Streaming about cleanSource https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#input-sources

1 Answers

Try using option("spark.sql.streaming.fileSource.cleaner.numThreads", "10") to speedup cleanup. If more files are getting generated in less time, then Spark don't delete. May be increasing threads helps

Related