Large dataset with pyspark - optimizing join, sort, compare between rows and group by with aggregation
I believe the below solution should solve for what you are looking to do in a more efficient way. Your current method involves a lot of "shuffle" operations (group by, sorting, joining). The below should help reduce the number of shuffle operation in your Spark job.


get leading start date
get disabled records
add column indicating whether product ever disabled (max of is disabled)
capture replacement dataset
get max replacement date
create indicator for current lifecycle records
filter data for current lifecycle records.




# csv input
df = spark.read.csv('productlist.csv', header=True, inferSchema=True)
# get ordered and unordered windows
wo = Window.partitionBy("product_id").orderBy("start_date")
wu = Window.partitionBy("product_id")
df1 = df.withColumn("lead_start_date", F.lead(col("start_date"), 1).over(wo))\
        .withColumn("is_disabled", F.when((col("end_date").isNotNull()) &amp;
                                          ((col("end_date") != col("lead_start_date")) | (col("lead_start_date").isNull())), 1).otherwise(0))\
        .withColumn("has_been_disabled", F.max(col("is_disabled")).over(wu))\
        .withColumn("replacement_date", F.when((col("end_date").isNotNull()) &amp;
                                          (col("end_date") == col("lead_start_date")) &amp; (col("lead_start_date").isNotNull()), col("start_date")).otherwise(lit(None)))\
        .withColumn("max_replacement_date", F.max(col("replacement_date")).over(wu))\
        .withColumn("is_current_lifecycle_record", F.when(((col("replacement_date") == col("max_replacement_date")) &amp; col("replacement_date").isNotNull()) |
                                                            ((col("has_been_disabled") == 0) &amp; (col("max_replacement_date").isNull())), 1).otherwise(0)) # never disabled / replaced
# filter for current lifecycle record and select target columns
df_final = df1.filter(col("is_current_lifecycle_record") == 1).select(["product_id", "start_date"])