Spark SQL - How to create marketing campaign paths for each user until purchase

user_15594687 0
Viewed 60

I am using Spark SQL to process the following dataset, so it can fit a marketing attribution model:

|   user_ID  |    timestamp     | activity |    campaign       | event_name
    ------        ---------        ------       --------           ------
| akalsds124 | 2021-12-31 10:00 | click    | Holidays Campaign | NULL
| akalsds124 | 2022-03-01 16:32 | click    | Super Campaign    | NULL 
| akalsds124 | 2022-04-27 20:55 | event    | NULL              | purchase
| akalsds124 | 2022-05-10 10:21 | event    | NULL              | purchase  
| akalsds124 | 2022-06-25 09:22 | click    | IG 3 Campaign     | NULL
| akalsds124 | 2022-07-07 15:00 | event    | NULL              | purchase
| ijnbmshs33 | 2022-05-02 10:31 | click    | New Campaign      | NULL
| ijnbmshs33 | 2022-07-04 17:01 | click    | Mega Campaign     | NULL

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

To create the table above, you can use this code:

df=spark.createDataFrame(
[('akalsds124','2021-12-31 10:00','click','Holidays Campaign','NULL'),
 ('akalsds124','2022-03-01 16:32','click','Super Campaign','NULL'),
 ('akalsds124','2022-04-27 20:55','event','NULL','purchase'),
 ('akalsds124','2022-05-10 10:21','event','NULL', 'purchase'),
 ('akalsds124','2022-06-25 09:22','click','IG 3 Campaign','NULL'),
 ('akalsds124','2022-07-07 15:00','event','NULL','purchase'),
 ('ijnbmshs33','2022-05-02 10:31','click','New Campaign','NULL'),
 ('ijnbmshs33','2022-07-04 17:01','click','Mega Campaign','NULL')],
 ['user_id','timestamp','activity','campaign','event_name']
)

I need to create a path with each user's campaign touchpoints inside a list. When a user purchases a product, a new path must be created for his/her next touchpoints.

Also, I need a column named 'converted' with boolean results (1 if the path led to a purchase and 0 if it did not lead to a conversion), and another one (total_conversions) with the total n° of purchases per path.

The expected output should be like this:

|  user_ID      |              path                   | converted | total_conversions 
    -----                     ------                      -----          ------- 
| akalsds124    | [Holidays Campaign,Super Campaign]  |    1      |        2
| akalsds124    | [IG Campaign]                       |    1      |        1
| ijnbmshs33    | [New Campaign,Mega Campaign]        |    0      |        0
pyspark apache-spark-sql apache-spark sql
1 Answers

Starting from the dataset you created, here is what i've done :

data preparation

from pyspark.sql import functions as F, Window as W

df = df.withColumn(
    "event_name", F.when(F.col("event_name") == "purchase", 1).otherwise(0)
)


df = df.withColumn(
    "rnk", F.lag("event_name").over(W.partitionBy("user_id").orderBy("timestamp"))
)


df = df.withColumn(
    "rnk", F.when((F.col("rnk") == 1) & (F.col("event_name") != 1), 1).otherwise(0)
)


df = df.withColumn(
    "rnk", F.sum("rnk").over(W.partitionBy("user_id").orderBy("timestamp"))
)

aggregation

df = df.groupBy("user_id", "rnk").agg(
    F.collect_set("campaign").alias("path"),
    F.max("event_name").alias("converted"),
    F.sum("event_name").alias("total_conversions"),
)

Result

df.show()

+----------+---+--------------------+---------+-----------------+
|   user_id|rnk|                path|converted|total_conversions|
+----------+---+--------------------+---------+-----------------+
|akalsds124|  0|[Super Campaign, ...|        1|                2|
|akalsds124|  1|[NULL, IG 3 Campa...|        1|                1|
|ijnbmshs33|  0|[Mega Campaign, N...|        0|                0|
+----------+---+--------------------+---------+-----------------+
user_50137520
Related