title: Preventing Spark from storing state in stream/stream joins
body: <p>I have two streaming datasets, let's call them <code>fastStream</code> and <code>slowStream</code>.</p>
  <p>The <code>fastStream</code> is a streaming dataset that I am consuming from Kafka via the structured streaming API. I am expecting to receive potentially thousands of messages a second.</p>
  <p>The <code>slowStream</code> is actually a reference (or lookup) table that is being 'upserted' by another stream and contains data that I want to join on to each message in the <code>fastStream</code> before I save the records to a table. The <code>slowStream</code> is only updated when someone changes the metadata, which can happen at any time but we would expect to change maybe once every few days.</p>
  <p>Each record in the <code>fastStream</code> will have exactly one corresponding message in the <code>slowStream</code> and I essentially want to make that join happen immediately with whatever data is in the <code>slowStream</code> table. I don't want to wait to see if a potential match could occur if new data arrives in the <code>slowStream</code>.</p>
  <p>The problem that I have is that according to the Spark docs:</p>
  <blockquote>
  <p>Hence, for both the input streams, we buffer past input as streaming state, so that we can match every future input with past input and accordingly generate joined results.</p>
  </blockquote>
  <p>I have tried adding a watermark to the <code>fastStream</code> but I think it has no effect since the docs indicate that the watermarked columns need to be referenced in the join</p>
  <p>Ideally I would write something like:</p>
  <pre class="lang-py prettyprint-override"><code># Apply a watermark to the fast stream
  fastStream = spark.readStream \
  .format(&quot;delta&quot;) \
  .load(&quot;dbfs:/mnt/some_file/fastStream&quot;) \
  .withWatermark(&quot;timestamp&quot;, &quot;1 hour&quot;) \
  .alias(&quot;fastStream&quot;)
  
  # The slowStream cannot be watermarked since it is only slowly changing
  slowStream = spark.readStream \
  .format(&quot;delta&quot;) \
  .load(&quot;dbfs:/mnt/some_file/slowStream&quot;) \
  .alias(&quot;slowStream&quot;)
  
  # Prevent the join from buffering the fast stream by 'telling' spark that there will never be new matches.
  fastStream.join( 
    slowStrean,
    expr(&quot;&quot;&quot; 
      fastStream.slow_id = slowStream.id
      AND fastStream.timestamp &gt; watermark
      &quot;&quot;&quot;
    ),
    &quot;inner&quot;
  ).select(&quot;fastStream.*&quot;, &quot;slowStream.metadata&quot;)
  
  </code></pre>
  <p>But I don't think you can reference the <code>watermark</code> in the SQL expression.</p>
  <p>Essentially, while I'm happy to have the <code>slowStream</code> buffered (so the whole table is in memory) I can't have the <code>fastStream</code> buffered as this table will quickly consume all memory. Instead, I would simply like to drop messages from the <code>fastStream</code> that aren't matched instead of retaining them to see if they might match in future.</p>
  <p>Any help very gratefully appreciated.</p>