title: structured streaming multiple streams reading from single kinesis streaming dataframe causes excessive kinesis GET requests (no watermarking)
body: <p>Here is my scenario:</p>
  <p>I have a single kinesis streaming dataframe getting json data published to it.</p>
  <p>I am trying to filter this json data on type by 3 streams &amp; write each type to 3 locations via these 3 streams.</p>
  <p>There is a 4th stream which uses the 1st type DF and joins 2 static dfs jdbc tables to it, writing it to a final jdbc table.</p>
  <pre><code>#Kinesis Streaming Dataframe
  KStr_DF = spark.readStream.format(&quot;kinesis&quot;).option(&quot;initialPosition&quot;, &quot;LATEST&quot;).load()
  #3 streaming dataframes filtered
  Str_DF_A = KStr_DF.where('requestData.type == &quot;typeA&quot;')
  Str_DF_B = KStr_DF.where('requestData.type == &quot;typeB&quot;')
  Str_DF_C = KStr_DF.where('requestData.type == &quot;typeC&quot;')
  
  Str_DF_A.writeStream()\
  .foreachBatch(lambda df,epochId:rows_writer(df,epochId,loc1,loc2, loc3))\
  .option(&quot;checkpointLocation&quot;,fpath1).start()
  Str_DF_B.writeStream()\
  .foreachBatch(lambda df,epochId:rows_writer(df,epochId,loc1,loc2, loc3))\
  .option(&quot;checkpointLocation&quot;,fpath2).start()
  Str_DF_C.writeStream()\
  .foreachBatch(lambda df,epochId:rows_writer(df,epochId,loc1,loc2, loc3))\
  .option(&quot;checkpointLocation&quot;,fpath3).start()
  
  #read static tables and join to first stream, then create 4th stream
  DFstaticD=spark.read.format(&quot;jdbc&quot;).load(tableD)
  DFstaticE=spark.read.format(&quot;jdbc&quot;).load(tableE) #jdbc DF placeholder
  
  #Stream Static joins
  StrStatTypeAD = Str_DF_A.join(DFstaticA,on='col1',type=&quot;left-outer&quot;)
  StrStatTypeADE = StrStatTypeAD.join(DFstaticE,on='col1',type=&quot;inner&quot;)
  
  #stream 4
  StrStatTypeADE.writeStream\
  .trigger(processingTime='10 seconds')\ #trigger of any use??
  .foreach(foreach_rows_writer(jdbc_url))\
  .option(&quot;checkpointLocation&quot;, fpath).outputMode(&quot;append&quot;).start()
  
  def rows_writer(batchdata, epochId, location1, location2, location3):
     batchdata.persist()
     batchdata.write().format(loc1).mode(&quot;append&quot;).save(location1)
     batchdata.write().format(loc2).mode(&quot;append&quot;).save(location2)
     batchdata.write().format(loc3).mode(&quot;append&quot;).save(location3)
     batchdata.unpersist()
  
  class foreach_rows_writer:
    #jdbc writes for inserts/updates to table D, left joined with Str_DF_A earlier
    def __init__(self,jdbc_url):
    def open(self, partition_id, epoch_id):
      #if connects return True else False
    def process(self, row):
      #cursor update or inserts 
  </code></pre>
  <p>My questions, for spark 3.1.x</p>
  <ol>
  <li>Basis <a href="https://stackoverflow.com/questions/57367226/structured-streaming-writing-to-multiple-streams">this</a>, all of the first 3 streams repoll/read from kinesis separately. Is it still true for 3.1.x? Or do they read from kinesis only once via the single streaming dataframe KStr_DF?</li>
  <li>The 4th stream, which uses the same dataframe that stream 1 uses and then joins it with other static dataframes, do the 1st/4th read/poll kinesis separately, or once only and use the same offsets via Str_DF_A?</li>
  <li>Can lack of watermarking on the joins cause excessive kinesis polling? Or if I were to dedupe the input stream kinesis on a column, without watermark?</li>
  <li>Is there any point using trigger(processingTime='10 seconds') with foreach(stream 4)?</li>
  </ol>