导入必要的包

from pyspark.sql.functions import col, when, sum, to_date

1.将日期从StringType转换为DateType

cast_date = lambda df, column: df.withColumn(column, to_date(column, "M-yyyy"))

df1 = df1 \
    .transform(cast_date, "dateM3") \
    .transform(cast_date, "dateM1") \
    .transform(cast_date, "date") \
    .transform(cast_date, "dateP3")

df2 = df2.transform(cast_date, "date")

1.使用条件语句联接并执行groupby求和

joined_df = df1.join(df2, "client")

condition1 = col("dateChng").between(col("dateM3"), col("dateM1"))
condition2 = col("dateChng").between(col("date"), col("dateP3"))

conditional_sum = lambda condition, col_name, col_alias: sum(when(condition, col(col_name)).otherwise(0)).alias(col_alias)

joined_df = joined_df.groupBy("client").agg(
    conditional_sum(condition1, "imp", "imp_range1"),
    conditional_sum(condition1, "con", "con_range1"),
    conditional_sum(condition2, "imp", "imp_range2"),
    conditional_sum(condition2, "con", "con_range2")
)

filter_cond = "imp_range1!=0 OR con_range1!=0 OR imp_range2!=0 OR con_range2!=0"

joined_df.filter(filter_cond).show()

输出：

+------+----------+----------+----------+----------+
|client|imp_range1|con_range1|imp_range2|con_range2|
+------+----------+----------+----------+----------+
|   123|         4|         1|         1|         0|
|   124|         0|         0|         3|         1|
+------+----------+----------+----------+----------+