Real-Time Recommendations Powered by Spanner, BigQuery, and Vector Embeddings

Product recommendation systems are an integral part of a wide range of industries like e-commerce, retail, media and entertainment, financial services, etc. Product recommendation is crucial for both providers and consumers as it improves the overall consumer experience and increases sales.

Businesses collect and analyze a ton of consumer usage and behavior data to optimize their  recommendations for purchase and user satisfaction. They strive to deliver these recommendations as soon as possible with the most up-to-date insights. Delays in showing relevant recommendations can result in lost sales and a bad experience for the consumer.

In this article, I would like to present a real-time recommendation system that operates on user and product vector embeddings and is powered by BigQuery (Google Cloud's fully managed, petabyte-scale data warehouse) and Spanner (Google Cloud's fully managed, mission-critical, global-scale database).

Vector Embeddings

Vector embeddings play a crucial role in generating recommendations. They capture user behavior, preferences, and intents based on their interaction with products or services. They also represent various features and attributes of a product or service. Vector embeddings do so by representing both users and products as high-dimensional numerical vectors. By computing distances between these vectors, we can measure similarities among products, among users, and between products and users.

Let's look at an example of vector embeddings. Below, we see two images: a baseball setup and a fishing setup. Feeding these images into Gemini and prompting something similar to "Generate a 64-dimensional vector embedding for these two images, keeping the dimensions the same for both embeddings" generates JSON arrays for two images with the same dimensions as described with the dimension labels in the images below.

Now, let's say that the user interacts with the baseball images first. As part of this interaction, we update the user embedding by aggregating the baseball embeddings with the user's existing embeddings (assuming the user embedding has the same dimensions). In this example, we aggregate with a simple average, but based on your business rules and requirements, this can be either a weighted average or any other complex function. After interacting with the fishing image, the system averages out the two embedding vectors and applies that to the user embedding. The updated vector is shown below.

Batch With BQ

Depending on the user traffic on the website or app, the amount of data flowing through the pipeline can be huge. Based on their interactions, most users may not need immediate product recommendations. We collect and batch-process these high-volume, high-velocity user interaction feeds in BigQuery with target (website and app elements like images, banners, buttons, etc.) embeddings. We then perform rolling aggregations with previously computed user embeddings and update the final user embeddings. These user embeddings are then pushed out to Spanner (via reverse ETL mechanisms) for real-time product recommendations for a given user ID.

Batch processing steps:

1. For the given batch duration, get distinct user IDs. This is to reduce the amount of data scanned from the user table in later steps.
2. For the given batch duration, join the events table with the targets table to map the target embeddings to individual user-target interactions.
3. Union all mapped embeddings with the respective user IDs from the users table.
4. Calculate the rolling average of the embeddings across each dimension for the respective users.
5. Append the updated embeddings to the users table.

---------------------------------------------------------------------------------------------
-- BQ Schema :
---------------------------------------------------------------------------------------------
-- dataset.events
--		user_id			String,
--		target_id		String,
-- 		ts			Timestamp

-- dataset.targets
--		target_id		String,
--		target_emb		String

-- dataset.users
--		user_id			String,
--		emb			String,
-- 		last_updated_ts		Timestamp
---------------------------------------------------------------------------------------------
WITH
  -- STEP 1
  dist_user_ids AS (
  SELECT
    DISTINCT events.user_id,
  FROM
    dataset.events
  WHERE
    events.ts >= $curr_batch_ts),
  -- STEP 2
  user_target_emb AS (
  SELECT
    events.user_id,
    1 AS target_count,
    targets.target_emb AS emb,
  FROM
    dataset.events events
  JOIN
    dataset.targets targets
  ON
    events.target_id = targets.target_id
  WHERE
    events.ts >= $curr_batch_ts
    -- STEP 3
  UNION ALL
  SELECT
    usres.user_id,
    users.target_count,
    users.emb,
  FROM
    dataset.users users
  JOIN
    dist_user_ids
  ON
    users.user_id = dist_user_ids.user_id),
  -- STEP 4
  emb_average AS (
  SELECT
    user_target_emb.user_id,
    idx,
    SUM(user_target_emb.target_count) AS target_count,
    SUM(user_target_emb.target_count * emb_val)/SUM(user_target_emb.target_count) new_emb_val
  FROM
    user_target_emb,
    UNNEST(user_target_emb.emb) emb_val
  WITH
  OFFSET
    AS idx
  GROUP BY
    1,
    2),
  updated_user_embeddings AS (
  SELECT
    user_id,
    ANY_VALUE(target_count) AS target_count,
    ARRAY_AGG(new_emb_val
    ORDER BY
      idx) AS new_emb
  FROM
    emb_average
  GROUP BY
    1 )
  -- STEP 5
SELECT
  user_id,
  target_count,
  new_emb AS emb,
  CURRENT_TIMESTAMP() AS last_updated_ts
FROM
  updated_user_embeddings;

Real-Time With Spanner

The updated user embeddings for the latest batch are pushed out to the corresponding Spanner table via Reverse ETL. The corresponding table for target embeddings is also maintained in Spanner. 

Event streams data in Spanner is needed only for the timestamps that are newer than the current batch being processed in BigQuery. We can schedule a job or assign TTL markers to periodically clean up this table.

In addition to the events, users, and targets tables, we also maintain the assets table, which holds the prediction assets for personalized recommendations. These assets have their own embeddings that match the dimensionality of the user and target embeddings.

When the prediction call comes in from the frontend for a given user:

1. All the latest events for that user are joined with the targets table to map the target embeddings to individual user-target interactions.
2. Union all mapped embeddings with the respective user IDs from the users table.
3. Calculate the final rolling average of the embeddings across each dimension for the given user.
4. The final user embedding is then used to calculate the distance from the assets.
5. The n closest assets are returned as part of the personalized recommendation prediction.

  ---------------------------------------------------------------------------------------------
  -- Spanner Schema :
  ---------------------------------------------------------------------------------------------
  -- events
  --		user_id			String,
  --		target_id		String,
  -- 		ts			Timestamp
  -- targets
  --		target_id		String,
  --		target_emb		String
  -- users
  --		user_id			String,
  --		emb			String,
  -- 		last_updated_ts		Timestamp
  -- assets
  --		asset_id		String,
  --		asset_emb		String,
  ---------------------------------------------------------------------------------------------
WITH
  -- STEP 1
  user_target_emb AS (
  SELECT
    events.user_id,
    1 AS target_count,
    targets.target_emb AS emb,
  FROM
    events
  JOIN
    targets
  ON
    events.target_id = targets.target_id
  WHERE
    events.user_id = "$user_id"
    -- STEP 2
  UNION ALL
  SELECT
    usres.user_id,
    users.target_count,
    users.emb,
  FROM
    users
  WHERE
    users.user_id = "$user_id" ),
  -- STEP 3
  emb_average AS (
  SELECT
    idx,
    SUM(user_target_emb.target_count * emb_val)/SUM(user_target_emb.target_count) new_emb_val
  FROM
    user_target_emb,
    UNNEST(user_target_emb.emb) emb_val
  WITH
  OFFSET
    AS idx
  GROUP BY
    1),
  updated_user_embeddings AS (
  SELECT
    ARRAY_AGG(new_emb_val
    ORDER BY
      idx) AS new_emb
  FROM
    emb_average ),
  -- STEP 4
  distances AS (
  SELECT
    asset_id,
    EUCLIDEAN_DISTANCE((
      SELECT
        new_emb
      FROM
        updated_user_embeddings),
      assets.asset_emb) AS distance,
  FROM
    assets)
  -- STEP 5
SELECT
  asset_id,
  distance
FROM
  distances
ORDER BY
  2 DESC
LIMIT
  $n

Conclusion

Based on the business requirements and guidelines, the front-end system will decide which product to show to the user as a final recommendation. This can be as simple as one with the shortest distance or a complex decision process with re-ranking of the predictions.

Putting it all together: 

As seen above, with this combination of real-time and batch processes, we take into account each and every user interaction and are able to recommend products and services that are relevant to the users in the context of their current 'space and time'!

This architecture is also elastic in nature and can scale depending on the user traffic and application requirements.