Hi everyone,

 

I've been trying to explore on cleanSource option in Autoloader to move files from the source to an archive location after they're processed and loaded into a table. I used the following simple code to test this functionality. While the code executes without any errors, the files remain in the source location and are not moved.

 

source_path = "abfss://container@storage_acc.dfs.core.windows.net/source/files"

archive_path = "abfss://container@storage_acc.dfs.core.windows.net/archive/files"

schema_location = "abfss://container@storage_acc.dfs.core.windows.net/source/autoloader/schema"

checkpoint_location = "abfss://container@storage_acc.dfs.core.windows.net/source/autoloader/checkpoint"

 

 

df = (

    spark.readStream

    .format("cloudFiles")

    .option("cloudFiles.format", "parquet")

    .option("cloudFiles.schemaLocation", schema_location)

    .option("cloudFiles.cleanSource", "MOVE")

    .option("cloudFiles.cleanSource.moveDestination", archive_path)

    .load(source_path))

)

 

 

df.writeStream

    .format("delta")

    .option("checkpointLocation", checkpoint_location)

    .outputMode("append")

    .table("uc.schema.table_name")

 

When I run the query SELECT * FROM cloud_files_state(checkpoint_location), I notice that the archive_mode and move_location columns are NULL, even though I’ve explicitly set cleanSource to MOVE. I also tested the DELETE option with .option("cloudFiles.cleanSource.retentionDuration", "7 days"), but that didn’t work either.

 

I came across a similar issue reported by another user using an S3 bucket as the source: Autoloader cleansource option does not take any effect 

 

I’ve tested this both in a notebook with a cluster running on Runtime 17.0 and using DLT with Runtime 16.4.

 

Could someone help me understand if I’m missing something or if there are any prerequisites or configurations needed to make this work?

 

Thanks,

Nikhil