Update marketing campaign paths based on time

Question

I am using Pyspark to process the following dataframe, so it can fit a marketing attribution model:

user_id	timestamp	activity	campaign	event_name
akalsds124	2022-01-01 10:00	click	Holidays Campaign	NULL
akalsds124	2021-12-31 09:00	click	Holidays Campaign	NULL
akalsds124	2022-01-13 15:59	click	X Campaign	NULL
akalsds124	2022-01-10 16:32	click	Super Campaign	NULL
akalsds124	2022-01-05 22:12	click	Holidays Campaign	NULL
akalsds124	2022-01-30 20:55	event	NULL	purchase
akalsds124	2022-01-30 22:10	event	NULL	purchase
akalsds124	2022-01-31 10:13	event	NULL	purchase
akalsds124	2022-02-03 04:55	click	T8 Campaign	NULL
akalsds124	2022-02-07 17:30	click	Y Campaign	NULL
akalsds124	2022-02-12 22:37	event	NULL	purchase
akalsds124	2022-03-31 18:19	click	U9 Campaign	NULL
akalsds124	2022-04-02 23:08	click	II Campaign	NULL
akalsds124	2022-03-02 07:00	click	T8 Campaign	NULL
ijnbmshs33	2022-06-03 17:01	click	Mega Campaign	NULL
ijnbmshs33	2022-05-03 10:31	click	New Campaign	NULL
ijnbmshs33	2022-05-20 17:01	click	Mega Campaign	NULL

An event is an interaction inside the app (e.g. a purchase, login, etc) and a click activity is an ad click made by the user.

I need to create a path with each user's campaign touchpoints inside a list. Each list must include only the touchpoints that the user interacted up to 30 days before the purchase (date of purchase has to be taken into account).

The paths that did not lead to a purchase must be updated after 30 days (the last day of the 30-day window must be counted). The order of the touchpoints is important and duplicates cannot be eliminated.

The output should be like this:

user_ID	path	converted	total_conversions
akalsds124	[Holidays Campaign,Holidays Campaign,Super Campaign,X Campaign]	1	2
akalsds124	[Holidays Campaign,Super Campaign,X Campaign]	1	1
akalsds124	[T8 Campaign, Y Campaign]	1	1
akalsds124	[T8 Campaign, U9 Campaign]	0	0
akalsds124	[II Campaign]	0	0
ijnbmshs33	[New Campaign,Mega Campaign]	0	0
ijnbmshs33	[Mega Campaign]	0	0

You can create the dataframe by using this code:

df=spark.createDataFrame(
        [('akalsds124','2022-01-01 10:00','click','Holidays Campaign','NULL'),
         ('akalsds124','2021-12-31 09:00','click','Holidays Campaign','NULL'),
         ('akalsds124','2022-01-13 15:59','click','X Campaign','NULL'),
         ('akalsds124','2022-01-10 16:32','click','Super Campaign','NULL'),
         ('akalsds124','2022-01-05 22:12','click','Holidays Campaign','NULL'),
         ('akalsds124','2022-01-30 20:55','event','NULL','purchase'),
         ('akalsds124','2022-01-30 22:10','event','NULL','purchase'),
         ('akalsds124','2022-01-31 10:13','event','NULL','purchase'),
         ('akalsds124','2022-02-03 04:55','click','T8 Campaign','NULL'),
         ('akalsds124','2022-02-07 17:30','click','Y Campaign','NULL'),
         ('akalsds124','2022-02-12 22:37','event','NULL','purchase'),
         ('akalsds124','2022-03-31 18:19','click','U9 Campaign','NULL'),
         ('akalsds124','2022-04-02 23:08','click','II Campaign','NULL'),
         ('akalsds124','2022-03-02 07:00','click','T8 Campaign','NULL'),
         ('ijnbmshs33','2022-06-03 17:01','click','Mega Campaign','NULL'),
         ('ijnbmshs33','2022-05-03 10:31','click','New Campaign','NULL'),
         ('ijnbmshs33','2022-05-20 17:01','click','Mega Campaign','NULL')],
         ['user_id','timestamp','activity','campaign','event_name']
    )

Dawyi · Answer

I think the trick is using collect_list function on a bounded window. the below code could be first part of your answer

window = W.partitionBy('user_id').orderBy('unixTime').rangeBetween(-3600*24*30, 0)
path_df = (
    df
    .withColumn('timestamp', F.col('timestamp').cast('timestamp'))
    .withColumn('unixTime', F.unix_timestamp('timestamp'))
    .withColumn('pathList', F.collect_list('campaign').over(window))
    .filter(F.col('event_name') == 'purchase')
)

path_df.sort('timestamp').show(truncate=False)

Update marketing campaign paths based on time

1 Answers