我需要一些关于以下情况的专家意见：
我有以下数据框df1：
+------------+------------+-------+-------+
| 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 |
+------------+------------+-------+-------+
每个group都是按OrderDate顺序排列的descending。订购后，每个value有都Current_date < (Date1 + 31Days) or Date1 as NULL需要标记为validuntil Current_date > (Date1 + 31Days)。发布后，每个Value都应标记为Invalid不考虑Date1价值。
如果对于 a group，所有的记录都是NULL，所有的Value都应该被标记为Valid
我的输出 df 应该如下所示：
+------------+------------+-------+-------+---------+
| 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 |
+------------+------------+-------+-------+---------+
我的做法：
我 在订购后row_number为每个创建。发布我得到每个的信息并将其保存为新的数据框。groupOrderDatemin(row_number)Current_date > (Date1 + 31Days)groupdfMin
然后我加入它 df1并dfMin基于grouprow_number(row_number < min(row_number))
这种方法适用于大多数情况。但是当对于 are 的所有group值时，这种方法会失败。Date1NULL
还有其他更好的方法来包含上述场景吗？
注意：我使用的是相当旧的 Spark- 版本Spark 1.5。windows功能在我的环境中也不起作用（它是一个自定义框架，并且有很多限制）。对于row_number，我使用zipWithIndex了方法。