Question: I have a dataframe with a status (integer) and a timestamp. Since I get a lot of "duplicated" status messages, I want to reduce the dataframe by removing any row which repeats a previous status within a 24h window after a "new" status, meaning:  The first 24h window starts with the first message of a specific status. The next 24h window for that status starts with the next message that comes after that first 24h window (the windows are not back-to-back).  Given the example: data = [(10, datetime.datetime.strptime("2022-01-01 00:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-01 04:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-01 23:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-02 05:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-02 06:00:00", "%Y-%m-%d %H:%M:%S")), (20, datetime.datetime.strptime("2022-01-01 03:00:00", "%Y-%m-%d %H:%M:%S")) ] myschema = StructType( [ StructField("status", IntegerType()), StructField("ts", TimestampType()) ] ) df = spark.createDataFrame(data=data, schema=myschema)   The first 24h window for status 10 is from 2022-01-01 00:00:00 until 2022-01-02 00:00:00. The second 24h window for status 10 is from 2022-01-02 05:00:00 until 2022-01-03 05:00:00. The first 24h window for status 20 is from 2022-01-01 03:00:00 until 2022-01-02 03:00:00.  As a result, I want to keep the messages: data = [(10, datetime.datetime.strptime("2022-01-01 00:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-02 05:00:00", "%Y-%m-%d %H:%M:%S")), (20, datetime.datetime.strptime("2022-01-01 03:00:00", "%Y-%m-%d %H:%M:%S")) ]  I know how to do this in Python by looping and keeping track of the latest change and I think I need to use a Window function with partitionBy + orderBy, but I cannot figure out the details... any help is appreciated.

Answer: That's a nice question, i spent some time trying to fiugre it out. I have something like this: At first i am calculating time diff in hours from prev element for each row within window Then i am aggregating above values together as a new column for each row (first row has null, next [4], third [4,19] etc Then i am using accumulator to calculate cumSum with your condition, so it "resets" after 24h Then i am using this cumSum to assign "inner window" Filtering at the end and i am done  Probably this is not the best way to do it so comments appreciated :D Sample code: import datetime from pyspark.sql import Window import pyspark.sql.functions as F data = [ (10, datetime.datetime.strptime("2022-01-01 00:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-01 04:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-01 23:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-02 05:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-02 06:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-02 06:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-02 16:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-03 16:00:00", "%Y-%m-%d %H:%M:%S")), (10, datetime.datetime.strptime("2022-01-03 17:00:00", "%Y-%m-%d %H:%M:%S")), (20, datetime.datetime.strptime("2022-01-01 03:00:00", "%Y-%m-%d %H:%M:%S")) ] myschema = StructType( [StructField("status", IntegerType()), StructField("ts", TimestampType())] ) df = spark.createDataFrame(data=data, schema=myschema) window = Window.partitionBy("status").orderBy(["ts"]) df = df.withColumn( "timeDiffFromPrev", (((F.col("ts")).cast("long") - F.lag("ts").over(window).cast("long")) / 3600).cast("long"), ) df = df.withColumn('aggTimeDiffFromPrev', F.collect_list('timeDiffFromPrev').over(window)) #Zero aggregate when sum > 24h expr = "AGGREGATE(aggTimeDiffFromPrev, 0l, (accumulator, element) -> IF(accumulator < 24, accumulator + element, element))" df = df.select('status', 'ts', F.expr(expr).alias('cumsum')) #Assign window number based on cumSum - anything greater than 24 is beggining of new window df = df.withColumn( "window_number", F.sum( F.when((F.col("cumsum") > 24), F.lit(1)).otherwise(F.lit(0)) ).over(window), ) innerWindow = Window.partitionBy("status", "window_number").orderBy(["ts"]) #Keep only first row in window, drop other df = df.withColumn("row_number", F.row_number().over(innerWindow)).filter( F.col("row_number") == F.lit(1) ).drop("timeDiffFromFirst", "window_number", "row_number", "cumSum") df.show()  I added few line on the input just for tests, for my input dataset results are as follows: +------+-------------------+ |status| ts| +------+-------------------+ | 10|2022-01-01 00:00:00| | 10|2022-01-02 05:00:00| | 10|2022-01-03 16:00:00| | 20|2022-01-01 03:00:00| +------+-------------------+