title: How can I control the amount of files being processed for each trigger in Spark Structured Streaming using the "Trigger once" trigger?
body: <p>I am trying to use Spark Structured Streaming's feature, Trigger once, to mimic a batch alike setup. However, I run into some trouble when I am running my initial batch, because I have a lot of historic data, and for this reason I am also using the option <strong>.option(&quot;cloudFiles.includeExistingFiles&quot;, &quot;true&quot;)</strong> to also process existing files.</p>
  <p>So my initial batch becomes very big, since I cannot control the amount of files for the batch.</p>
  <p>I have also tried to use the option <strong>cloudFiles.maxBytesPerTrigger</strong>, however, this is ignored when you use Trigger once --&gt; <a href="https://docs.databricks.com/spark/latest/structured-streaming/auto-loader-gen2.html" rel="nofollow noreferrer">https://docs.databricks.com/spark/latest/structured-streaming/auto-loader-gen2.html</a></p>
  <p>When I specify the <strong>maxFilesPerTrigger</strong> option, it is ignored as well. It just takes all files available.</p>
  <p>My code looks like this:</p>
  <pre><code>df = (
    spark.readStream.format(&quot;cloudFiles&quot;)
      .schema(schemaAsStruct)
      .option(&quot;cloudFiles.format&quot;, sourceFormat)
      .option(&quot;delimiter&quot;, delimiter)
      .option(&quot;header&quot;, sourceFirstRowIsHeader)
      .option(&quot;cloudFiles.useNotifications&quot;, &quot;true&quot;)
      .option(&quot;cloudFiles.includeExistingFiles&quot;, &quot;true&quot;)
      .option(&quot;badRecordsPath&quot;, badRecordsPath)
      .option(&quot;maxFilesPerTrigger&quot;, 1)
      .option(&quot;cloudFiles.resourceGroup&quot;, omitted)
      .option(&quot;cloudFiles.region&quot;, omitted)
      .option(&quot;cloudFiles.connectionString&quot;, omitted)
      .option(&quot;cloudFiles.subscriptionId&quot;, omitted)
      .option(&quot;cloudFiles.tenantId&quot;, omitted)
      .option(&quot;cloudFiles.clientId&quot;, omitted)
      .option(&quot;cloudFiles.clientSecret&quot;, omitted)
      .load(sourceBasePath)
  )
  
  # Traceability columns
  df = (
    df.withColumn(sourceFilenameColumnName, input_file_name()) 
      .withColumn(processedTimestampColumnName, lit(processedTimestamp))
      .withColumn(batchIdColumnName, lit(batchId))
  )
  
  def process_batch(batchDF, id):
    batchDF.persist()
    
    (batchDF
       .write
       .format(destinationFormat)
       .mode(&quot;append&quot;)
       .save(destinationBasePath + processedTimestampColumnName + &quot;=&quot; +  processedTimestamp)
    )
      
    (batchDF
     .groupBy(sourceFilenameColumnName, processedTimestampColumnName)
     .count()
     .write
     .format(destinationFormat)
     .mode(&quot;append&quot;)
     .save(batchSourceFilenamesTmpDir))
    
    batchDF.unpersist()
  
  stream = (
    df.writeStream
      .foreachBatch(process_batch)
      .trigger(once=True)
      .option(&quot;checkpointLocation&quot;, checkpointPath)
      .start()
  )
  </code></pre>
  <p>As you can see, I am using the <strong>cloudfiles</strong> format, which is the format of the Databricks Autoloader --&gt; <a href="https://docs.databricks.com/spark/latest/structured-streaming/auto-loader-gen2.html" rel="nofollow noreferrer">https://docs.databricks.com/spark/latest/structured-streaming/auto-loader-gen2.html</a></p>
  <p>&quot;<em>Auto Loader incrementally and efficiently processes new data files as they arrive in cloud storage.</em></p>
  <p><em>Auto Loader provides a Structured Streaming source called cloudFiles. Given an input directory path on the cloud file storage, the cloudFiles source automatically processes new files as they arrive, with the option of also processing existing files in that directory</em>&quot;</p>
  <p>If I somehow present my problem in a confusing way or it's lacking information, please say so.</p>