我有三列的sparkDataframe。

final = spark.createDataFrame([[datetime(2021,1,1),None,1,1,1], 
[datetime(2021,1,2),5,None,3,1],[datetime(2021,1,3),2,None,4,1], 
[datetime(2021,1,4),3,None,3,1],[datetime(2021,1,5),2,1,None,1], 
[datetime(2021,1,6),1,8,None,1], [datetime(2021,1,7),2,3,5,1], 
[datetime(2021,1,1),None,1,1,2], [datetime(2021,1,2),5,None,3,2], 
[datetime(2021,1,3),2,None,4,2], [datetime(2021,1,4),3,None,3,2], 
[datetime(2021,1,5),2,1,None,2], [datetime(2021,1,6),1,8,None,2], 
[datetime(2021,1,7),2,3,5,2]]).toDF("a","b","c","d","e")

只有当两个相邻的非空值之间的间隙不大于2时，我才需要向前填充缺失的值，并对e列中的每个唯一记录应用相同的函数索引列将是a列
如何对pysparkDataframe执行此操作？