在询问Spark问题时，提供示例输入 Dataframe 是非常重要的，你没有提供，所以我假设你的输入 Dataframe 看起来像这样：

from pyspark.sql import functions as F, Window as W
df = spark.createDataFrame(
    [('1', '01',  100, 2),
     ('1', '02', None, 3),
     ('1', '03', None, 4),
     ('1', '04', None, 2)],
    ['id', 'month', 'value', 'monthly_increment'])

Spark3.2+**
您可以使用product、lag和first窗口函数的组合来填充缺少的列“值”值：

w = W.partitionBy('id').orderBy('month')
factor = F.product(F.lag('monthly_increment').over(w)).over(w)
df = df.withColumn('value', F.coalesce(F.first('value').over(w) * factor, 'value'))

df.show()

# +---+-----+------+-----------------+

# | id|month| value|monthly_increment|

# +---+-----+------+-----------------+

# |  1|   01| 100.0|                2|

# |  1|   02| 200.0|                3|

# |  1|   03| 600.0|                4|

# |  1|   04|2400.0|                2|

# +---+-----+------+-----------------+