batch process autoloader

seefoods — Thu, 03 Jul 2025 12:03:04 GMT

My job continue to running after is finished susccessfully this i my case, i enable useNotification

if self.autoloader_config.use_autoloader:
    logger_file_ingestion.info("debut d'ecriture en mode streaming")

    if self.write_mode.value.lower() == "append":
        logger_file_ingestion.info("ecriture en mode %s", self.write_mode.value)

        # Création de la configuration de base du stream
        stream_writer = (df.writeStream
                         .format("delta")
                         .outputMode("append")
                         .option("checkpointLocation", self.checkpoint_location)
                         .option("mergeSchema", "true")
                         .trigger(availableNow=True))

        # Ajout des partitions si nécessaire
        if (self.source_name.lower() == "name") and (self.file_format.lower() == "parquet"😞
            stream_writer = stream_writer.partitionBy("year", "day", "month")
        elif (self.source_name.lower() == "test") and (self.file_format.lower() == "parquet"😞
            stream_writer = stream_writer.partitionBy("day", "month", "year")

        # Lancement du stream et capture de la référence
        stream_writer.toTable("bronze")

Re: batch process autoloader

MariuszK — Fri, 04 Jul 2025 09:18:59 GMT

Did you try use .awaitTermination() on stream writing?

Re: batch process autoloader

seefoods — Tue, 08 Jul 2025 07:09:08 GMT

Hi @MariuszK,
Thanks it works now. i have use awaitTermination()

Re: batch process autoloader

MariuszK — Tue, 08 Jul 2025 08:07:29 GMT

Hi @seefoods ,
If it works, you can mark my answer as a solution so that if someone has the same problem, it will be easier to find an answer.

topic Re: batch process autoloader in Data Engineering

batch process autoloader

Re: batch process autoloader

Re: batch process autoloader

Re: batch process autoloader