Pyspark label rows based on time window and double condition match

You can use rangeBetween window function to look up last 7 days.

First cast datetime to long value to be able to used with rangeBetween.

# If the datetime is string
df = df.withColumn('datetime', F.to_timestamp(df.datetime, 'dd-MM-yy H:mm'))
df = df.withColumn('long_dt', df.datetime.cast(LongType()))

# If the datetime is timestamp type
df = df.withColumn('long_dt', df.datetime.cast(LongType()))

Then create window function to look up past 7 days with for the same user_ID and item.

w = (Window.partitionBy(['user_ID', 'item'])
    .orderBy('long_dt')
    .rangeBetween(-7*24*60*60, -1)) # -7 days to -1 second before.

Use this window function to count the row and if you see for the second time (> 0), you want to return 1. You also want to return only when online == no, so add the filter at the end.

Edit: Thanks to @blackbishop, I updated the counting logic.

df = (df.withColumn('result', (F.count(F.when(df.online == 'yes', 1)).over(w) > 0).cast(IntegerType()))
     .filter(df.online == 'no'))

Result

+-------+----+------+-------------------+------+
|user_id|item|online|           datetime|result|
+-------+----+------+-------------------+------+
|      1|  66|    no|2021-10-06 09:30:00|     1|
|      3|  31|    no|2021-10-15 12:07:00|     0|
|      3|  66|    no|2021-10-14 12:07:00|     1|
|      2|  24|    no|2021-10-03 08:01:00|     0|
|      2|  24|    no|2021-10-03 08:03:00|     1|
+-------+----+------+-------------------+------+

Other answers demonstrated how to use window functions. Here's a another way using left_semi join, which returns only users who bought items offline after they tried to bought the same items online 7 days earlier:

from pyspark.sql import functions as F

# cast string into timestamp if not already done
df = df.withColumn('datetime', F.to_timestamp("datetime", 'dd-MM-yy H:mm'))

df_on = df.filter("online = 'yes'").alias("on")
df_off = df.filter("online = 'no'").alias("off")

result = df_off.join(
    df_on, [
        F.col("on.user_ID") == F.col("off.user_ID"),
        F.col("on.item") == F.col("off.item"),
        F.col("on.datetime") >= F.expr("off.datetime - interval 7 day"),
        F.col("on.datetime") < F.col("off.datetime")
    ],
    "left_semi"
)

result.show()
#+-------+----+------+-------------------+
#|user_ID|item|online|           datetime|
#+-------+----+------+-------------------+
#|      1|  66|    no|2021-10-06 09:30:00|
#|      2|  24|    no|2021-10-03 08:03:00|
#|      3|  66|    no|2021-10-14 12:07:00|
#+-------+----+------+-------------------+

Note if you want the users who bought item offline but haven't tried to bought it online before (corresponding to result=0, replace semi join with left_anti in the above query.

I've included the pandas code to easily build out a sample dataframe. The spark code starts after that.

It's based upon groupby and shifting the online and datetime values. The idea being that the previous value to a no for any given id/item needs to be within 7 days to count.

import pandas as pd

from pyspark.sql.functions import lag, col, datediff, when
from pyspark.sql.window import Window

df = pd.DataFrame({'user_ID':[1,1,1,2,2,2,3,3,3,3],
                  'item':[66,24,66,24,24,24,31,66,66,31],
                   'online':['yes','yes','no','no','yes','no','yes','yes','no','no'],
                  'datetime':['02-10-21 9:30',
'02-10-21 9:30',
'06-10-21 9:30',
'03-10-21 8:01',
'03-10-21 8:02',
'03-10-21 8:03',
'05-10-21 8:32',
'10-10-21 10:57',
'14-10-21 12:07',
'15-10-21 12:07']})


df['datetime']  = pd.to_datetime(df['datetime'], dayfirst=True)


df = spark.createDataFrame(df)


df = df.withColumn('o_shift', lag(df['online']).over(Window.partitionBy('user_ID','item').orderBy('datetime')))
df = df.withColumn('d_shift', lag(df['datetime']).over(Window.partitionBy('user_ID','item').orderBy('datetime')))
df = df.dropna()

df = df.withColumn('t_delta', datediff(col('datetime'),col('d_shift')))


df = df.withColumn('result',when(
  (col('online')=='no') &
  (col('o_shift')=='yes') & 
  (col('t_delta')<=7),1
).otherwise(0))

df.sort('user_id','datetime').select('user_id','item','online','datetime','result').show()

Output

+-------+----+------+-------------------+------+
|user_id|item|online|           datetime|result|
+-------+----+------+-------------------+------+
|      1|  66|    no|2021-10-06 09:30:00|     1|
|      2|  24|   yes|2021-10-03 08:02:00|     0|
|      2|  24|    no|2021-10-03 08:03:00|     1|
|      3|  66|    no|2021-10-14 12:07:00|     1|
|      3|  31|    no|2021-10-15 12:07:00|     0|
+-------+----+------+-------------------+------+