title: SparkSteaming reading entire table instead of by file
body: <p>I have ~3PB of parquet on S3. I want to read it file-by-file with spark streaming and join some metadata to it before writing out. The metadata is small enough to be broadcasted. Files in the source data are ~60mb, none are huge.</p>
  <pre class="lang-scala prettyprint-override"><code>val r = spark.readStream
    .option(&quot;maxFilesPerTrigger&quot;, &quot;100&quot;)
    .schema(pschema)
    .parquet(&quot;s3://mybigdata/sourcedata/&quot;)
    .withColumn(&quot;id&quot;, regexp_extract(col(&quot;mycol&quot;), &quot;someregex&quot;, 1).cast(IntegerType))
    .alias(&quot;p&quot;)
    .join(broadcast(idmap.alias(&quot;i&quot;)), $&quot;p.id&quot; === $&quot;i.id&quot;, &quot;inner&quot;) //idmap is a small dataframe
    .drop($&quot;i.id&quot;)
    .withColumn(&quot;date&quot;, regexp_extract($&quot;filename&quot;, &quot;someregex&quot;, 1))
    
  val w = r.writeStream.format(&quot;delta&quot;)
    .partitionBy(&quot;date&quot;, &quot;some_id&quot;)
    .option(&quot;checkpointLocation&quot;, &quot;s3://mybigdata/checkpoint/&quot;)
    .option(&quot;path&quot;, &quot;s3://mybigdata/destination/&quot;)
    .start()
  </code></pre>
  <p>When I do this, I get MASSIVE spills to memory and disk:
  <a href="https://i.stack.imgur.com/dOjwh.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/dOjwh.png" alt="disk spills" /></a></p>
  <p>Which of course, is a disaster. How is it that I am getting these massive spills when I'm rate limiting via <code>maxFilesPerTrigger</code> to 100x60mb files at a time? It seems to be trying to read the <em>entire</em> S3 dataset and isn't streaming at all.</p>
  <p>What is going wrong here?</p>