Use RisingWave to Process and Transform Abandoned Cart Events

Overview

Abandoned cart data refers to the data collected when customers begin making purchases on an e-commerce platform by adding items to their shopping carts but do not complete the transaction. This is a valuable resource that businesses can leverage to improve their e-commerce strategies and customer retention rates. By knowing when a customer abandons their purchase, a business can set up real-time notifications to remind the customer to return to their cart and finish the purchase. Or, the business can offer the customer personalized discounts based on the items left in the cart to improve conversion rates.

To implement real-time capabilities or to perform real-time data transformations, a strong technological infrastructure is necessary. It may also be necessary to join existing batch data with data streams. This can be difficult to set up and maintain as there are numerous moving parts and many real-time data transformation tools are tricky to manage and use. However, RisingWave makes it simple to build real-time data pipelines. Both streaming data and batch data can be ingested into RisingWave, joined, transformed, and aggregated in real-time using SQL queries, and sinked to data warehouses and lakes. To see how easily you can run RisingWave, see Quick start.

This topic will introduce the idea of using RisingWave to process and transform abandoned cart data from multiple sources.

Step 1: Ingest data in RisingWave

In order to analyze abandoned cart data, we will process both real-time data from Kafka topics, and CDC data from MySQL. With RisingWave, we can easily ingest data from numerous sources with SQL queries.

MySQL data

Let us start by ingesting data from two MySQL tables. The first table stores all user data and the second stores all product data for the business. Their schemas are shown below.

-- User schema in MySQL
CREATE TABLE users (
  id INT PRIMARY KEY,
  first_name VARCHAR(50),
  last_name VARCHAR(50),
  email VARCHAR(100) UNIQUE
);

-- Product schema in MySQL
CREATE TABLE products (
  id INT PRIMARY KEY,
  name VARCHAR(100),
  description TEXT,
  price DECIMAL(10,2),
  category VARCHAR(50)
);

To read data from these two tables into RisingWave, we can use the following queries. For details on the syntax and how to set up MySQL, see Ingest data from MySQL CDC. As the source tables in MySQL are updated with new data entries, the sources in RisingWave will also instantaneously update.

-- Create the user table in RisingWave with MySQL as the source
CREATE TABLE users (
  id INT PRIMARY KEY,
  first_name VARCHAR(50),
  last_name VARCHAR(50),
  email VARCHAR(100)
) WITH (
  connector = 'mysql-cdc',
  hostname = '127.0.0.1',
  port = '3306',
  uesrname = 'your_username',
  password = 'your_password',
  database.name = 'mydb',
  table.name = 'users',
  server.id = '5454'
);

-- Create the products table in RisingWave with MySQL as the source
CREATE TABLE products (
  id INT PRIMARY KEY,
  name VARCHAR(100),
  description TEXT,
  price DECIMAL(10,2),
  category VARCHAR(50)
) WITH (
  connector = 'mysql-cdc',
  hostname = '127.0.0.1',
  port = '3306',
  uesrname = 'your_username',
  password = 'your_password',
  database.name = 'mydb',
  table.name = 'products',
  server.id = '5454'
);

Kafka topics

As shoppers are updating their carts, being able to instantly receive and process that data allows us to extract valuable insights in a timely manner and immediately take action, if necessary. By connecting RisingWave to a Kafka broker, we can stream these events into RisingWave in real-time. The schema of an abandoned cart event is shown below.

-- Topic: abandoned_carts
{
  "cart_id": INT,
  "user_id": INT,
  "product_id": INT,
  "timestamp": TIMESTAMP
}

With the following query, RisingWave allows us to easily connect to a Kafka broker to read data. For specifics on the query syntax and how to specify security settings, see Ingest data from Kafka. If you want to set up Kafka clusters on Confluent Cloud, see Ingest data from Confluent Cloud.

-- Define the abandoned cart stream in RisingWave
CREATE TABLE abandoned_carts (
  cart_id INT,
  user_id INT,
  product_id INT,
  timestamp TIMESTAMP
) WITH (
  KAFKA_BOOTSTRAP_SERVERS = 'kafka:9092',
  KAFKA_TOPIC = 'abandoned_carts'
);

Step 2: Join tables by creating a materialized view

To consolidate the data we just ingested, we will join the tables above by creating a materialized view. This materialized view will provide a clear overall picture of what events are occurring on the e-commerce platform. The benefit of creating a materialized view is that the results will immediately update as new data is ingested. For more on the syntax of creating a materialized view, see [CREATE MATERIALIZED VIEW].

The following query joins relevant fields from the three tables created in Step 1 by using multiple joins. For more on what type of joins RisingWave supports, see Joins.

CREATE MATERIALIZED VIEW abandoned_cart_analysis AS
SELECT
  a.cart_id,
  a.timestamp AS abandonment_time,
  u.id AS user_id,
  u.first_name,
  u.last_name,
  u.email,
  p.id AS product_id,
  p.name AS product_name,
  p.description AS product_description,
  p.price,
  p.category
FROM
  abandoned_carts a
  JOIN
    users u ON a.user_id = u.id
  JOIN
    products p ON a.product_id = p.id;

Here is an example of what the output might be if we query from abandoned_cart_analysis.

SELECT * FROM abandoned_cart_analysis LIMIT 5;

cart_id |  abandonment_time   | user_id | first_name | last_name |          email          | product_id | product_name | product_description       | price  | category   
--------+---------------------+---------+------------+-----------+-------------------------+------------+------------- +---------------------------+--------+------------
      1 | 2023-10-25 10:15:00 |       1 | John       | Doe       | johndoe@example.com     |        101 | Product A    | Description of Product A  |  19.99 | Category 1
      2 | 2023-10-25 11:30:00 |       2 | Alice      | Smith     | alicesmith@example.com  |        102 | Product B    | Description of Product B  |  29.99 | Category 2
      3 | 2023-10-25 12:45:00 |       3 | Bob        | Johnson   | bobjohnson@example.com  |        103 | Product C    | Description of Product C  |   9.99 | Category 1
      4 | 2023-10-25 14:00:00 |       4 | Eve        | Williams  | evewilliams@example.com |        104 | Product D    | Description of Product D  |  39.99 | Category 3
      5 | 2023-10-26 15:15:00 |       1 | John       | Doe       | johndoe@example.com     |        102 | Product B    | Description of Product B  |  29.99 | Category 2

Step 3: Further analysis

Using the abandoned_cart_analysis materialized view, we can perform numerous types of analysis related to abandoned shopping carts, users, and products. We can either query directly from the materialized view to get an overall idea during a specific moment or build materialized views on top to create live dashboards.

For instance, we can analyze users' behaviors by identifying users with the most abandoned carts with the following SQL query. By creating a materialized view, these results will be constantly updated, allowing us to see which users are often browsing the catalog, but not purchasing many items.

CREATE MATERIALIZED VIEW most_abandoned AS
SELECT user_id, COUNT(DISTINCT cart_id) AS abandoned_carts
FROM abandoned_cart_analysis
GROUP BY user_id;

If we query from the materialized view and order abandoned_carts from greatest to least, we can see which customers are not making many purchases. Note that the ORDER BY clause was not included when creating the materialized view. The order is not maintained as the materialized view gets refreshed with new data. The results may look like the following.

SELECT * FROM most_abandoned
ORDER BY abandoned_carts DESC;

 user_id | abandoned_carts
---------+-----------------
 39      | 40
 10      | 23
 439     | 12
 219     | 4

This is just a simple example of the type of analysis you can perform based on the abandon_cart_analysismaterialized view. RisingWave supports many SQL commands and functions that you can leverage to conduct a more detailed and thorough analysis.