title: How to pick latest record in spark structured streaming join
body: <p>I am using spark-sql 2.4.x version , datastax-spark-cassandra-connector for Cassandra-3.x version. Along with kafka.</p>
  
  <blockquote>
    <p>I have rates meta data of currency sample as below :</p>
  </blockquote>
  
  <pre><code>val ratesMetaDataDf = Seq(
       ("EUR","5/10/2019","1.130657","USD"),
       ("EUR","5/9/2019","1.13088","USD")
       ).toDF("base_code", "rate_date","rate_value","target_code")
  .withColumn("rate_date", to_date($"rate_date" ,"MM/dd/yyyy").cast(DateType))
  .withColumn("rate_value", $"rate_value".cast(DoubleType))
  </code></pre>
  
  <blockquote>
    <p>Sales records which i received from kafka topic is , as (sample) below
    :</p>
  </blockquote>
  
  <pre><code>val kafkaDf = Seq((15,2016, 4, 100.5,"USD","2021-01-20","EUR",221.4)
                                  ).toDF("companyId", "year","quarter","sales","code","calc_date","c_code","prev_sales")
  </code></pre>
  
  <p>To calculate "prev_sales" , I need get its "c_code" 's respective "rate_value" which is nearest to the "calc_date" i.e. rate_date"</p>
  
  <p>Which i am doing as following</p>
  
  <pre><code>val w2 = Window.orderBy(col("rate_date") desc)
  val rateJoinResultDf = kafkaDf.as("k").join(ratesMetaDataDf.as("e"))
                                     .where( ($"k.c_code" === $"e.base_code") &amp;&amp;
                                             ($"rate_date" &lt; $"calc_date")
                                           ).orderBy($"rate_date" desc)
                                    .withColumn("row",row_number.over(w2))
                                    .where($"row" === 1).drop("row")
                                    .withColumn("prev_sales", (col("prev_sales") * col("rate_value")).cast(DoubleType))
                                    .select("companyId", "year","quarter","sales","code","calc_date","prev_sales")
  </code></pre>
  
  <p>In the above to get nearest record (i.e. "5/10/2019" from ratesMetaDataDf  ) for given "rate_date" I am using window and row_number function and sorting the records by "desc".</p>
  
  <blockquote>
    <p>But in the spark-sql streaming it is causing the error as below</p>
  </blockquote>
  
  <pre><code>"
  Sorting is not supported on streaming DataFrames/Datasets, unless it is on aggregated DataFrame/Dataset in Complete output mode;;"
  </code></pre>
  
  <p>So how to fetch first record to join in the above.</p>