Pyspark: как применить к значению фрейма данных другое значение в зависимости от даты в другом фрейме данных

Question

У меня есть мой первый фрейм данных df, который содержит start_date и значение, и мой второй фрейм данных df_v, который содержит только даты.

Мой df:

+-------------------+-----+
|      start_date   |value|
+-------------------+-----+
|2019-03-17 00:00:00|   35|
+-------------------+-----+
|2019-05-20 00:00:00|   40|
+-------------------+-----+
|2019-06-03 00:00:00|   10|
+-------------------+-----+
|2019-07-01 00:00:00|   12|
+-------------------+-----+

мой df_v:

+-------------------+
|       date        |
+-------------------+
|2019-02-01 00:00:00|
+-------------------+
|2019-04-10 00:00:00|
+-------------------+
|2019-06-14 00:00:00|   
+-------------------+

Я хочу новый df_v:

+-------------------+-------------+
|       date        |   v_value   |
+-------------------+-------------+
|2019-02-01 00:00:00|            0|
+-------------------+-------------+
|2019-04-10 00:00:00|    (0+35) 35|
+-------------------+-------------+
|2019-06-14 00:00:00|(35+40+10) 85|
+-------------------+-------------+

пытается работать так:

df=df.withColumn("lead",lead(F.col("start_date"),1).over(Window.orderBy("start_date")))

for r_v in df_v.rdd.collect():
    for r in df.rdd.collect():
        if (r_v.date >= r.start_date) and (r_v.date < r.lead):
            df_v = df_v.withColumn('v_value', 
            ...

Answer 1

Это можно сделать с помощью join и агрегации.

from pyspark.sql.functions import sum,when
#Join
joined_df = df_v.join(df,df.start_date <= df_v.date,'left')
joined_df.show() #View the joined result 
#Aggregation
joined_df \
.groupBy(joined_df.date) \
.agg(sum((when(joined_df.value.isNull(),0).otherwise(joined_df.value))).alias('val')) \
.show()