title: Flagging records after meeting a condition using Spark Scala
body: <p>I need some expert opinion on the below scenario:</p>
  <p>I have following dataframe <code>df1</code>:</p>
  <pre><code>+------------+------------+-------+-------+
  |   Date1    | OrderDate  | Value | group |
  +------------+------------+-------+-------+
  | 10/10/2020 | 10/01/2020 | hostA | grp1  |
  | 10/01/2020 | 09/30/2020 | hostB | grp1  |
  | Null       | 09/15/2020 | hostC | grp1  |
  | 08/01/2020 | 08/30/2020 | hostD | grp1  |
  | Null       | 10/01/2020 | hostP | grp2  |
  | Null       | 09/28/2020 | hostQ | grp2  |
  | 07/11/2020 | 08/08/2020 | hostR | grp2  |
  | 07/01/2020 | 08/01/2020 | hostS | grp2  |
  | NULL       | 07/01/2020 | hostL | grp2  |
  | NULL       | 08/08/2020 | hostM | grp3  |
  | NULL       | 08/01/2020 | hostN | grp3  |
  | NULL       | 07/01/2020 | hostO | grp3  |
  +------------+------------+-------+-------+
  </code></pre>
  <p>Each <code>group</code> is ordered by <code>OrderDate</code> in <code>descending</code> order. Post ordering, Each <code>value</code> having <code>Current_date &lt; (Date1 + 31Days) or Date1 as NULL</code> needs to be flagged as <code>valid</code> until <code>Current_date &gt; (Date1 + 31Days)</code>.
  Post that, every <code>Value</code> should be marked as <code>Invalid</code> irrespective of <code>Date1</code> value.</p>
  <p>If for a <code>group</code>, all the records are <code>NULL</code>, all the <code>Value</code> should be tagged as <code>Valid</code></p>
  <p>My output df should look like below:</p>
  <pre><code>+------------+------------+-------+-------+---------+
  |   Date1    | OrderDate  | Value | group |  Flag   |
  +------------+------------+-------+-------+---------+
  | 10/10/2020 | 10/01/2020 | hostA | grp1  | Valid   |
  | 10/01/2020 | 09/30/2020 | hostB | grp1  | Valid   |
  | Null       | 09/15/2020 | hostC | grp1  | Valid   |
  | 08/01/2020 | 08/30/2020 | hostD | grp1  | Invalid |
  | Null       | 10/01/2020 | hostP | grp2  | Valid   |
  | Null       | 09/28/2020 | hostQ | grp2  | Valid   |
  | 07/11/2020 | 08/08/2020 | hostR | grp2  | Invalid |
  | 07/01/2020 | 08/01/2020 | hostS | grp2  | Invalid |
  | NULL       | 07/01/2020 | hostL | grp2  | Invalid |
  | NULL       | 08/08/2020 | hostM | grp3  | Valid   |
  | NULL       | 08/01/2020 | hostN | grp3  | Valid   |
  | NULL       | 07/01/2020 | hostO | grp3  | Valid   |
  +------------+------------+-------+-------+---------+
  </code></pre>
  <p><strong>My approach:</strong></p>
  <p>I created  <code>row_number</code> for each <code>group</code> after ordering by <code>OrderDate</code>.
  Post that i am getting the <code>min(row_number)</code> having <code>Current_date &gt; (Date1 + 31Days)</code> for each <code>group</code> and save it as new dataframe <code>dfMin</code>.</p>
  <p>I then join it with  <code>df1</code> and <code>dfMin</code> on <code>group</code> and filter based on <code>row_number(row_number &lt; min(row_number))</code></p>
  <p>This approach works for most cases. But when for a <code>group</code> all values of <code>Date1</code> are <code>NULL</code>, this approach fails.</p>
  <p>Is there any other better approach to include the above scenario as well?</p>
  <p><em><strong>Note:</strong> I am using pretty old version of Spark- <code>Spark 1.5</code>. Also <code>windows</code> function <strong>won't work</strong> in my environment(Its a custom framework and there are many restrictions in place). For <code>row_number</code>, i used <code>zipWithIndex</code> method.</em></p>