Build recommendation systems with FeatureStore Python SDK

Synchronize data tables

A recommendation scenario typically requires the following data tables: a user feature table, an item table, a label table, a sequence feature table, and a behavior table.

This hands-on tutorial uses simulated data tables prepared in the pai_online_project MaxCompute project. To use them, run SQL commands in DataWorks to synchronize the tables to your own MaxCompute project. Follow these steps:

1. Log on to the DataWorks console.

2. In the navigation pane on the left, click Data Development & O&M > Data Development.

3. Select your DataWorks workspace and click Enter Data Development.

4. Hover over New, and then choose New Node > MaxCompute > ODPS SQL. Configure the node parameters as follows:

   Parameter	Recommended value
   engine instance	Select the MaxCompute engine that you created.
   node type	ODPS SQL
   path	Business Flow/Workflow/MaxCompute
   name	Enter a custom name.

5. Click Confirm.

6. In the node editor, run the following SQL commands to synchronize the data tables to your MaxCompute project. For Resource Group, select your exclusive resource group.

   User table: rec_sln_demo_user_table_preprocess_all_feature_v1

   CREATE TABLE IF NOT EXISTS rec_sln_demo_user_table_preprocess_all_feature_v1
   like pai_online_project.rec_sln_demo_user_table_preprocess_all_feature_v1
   STORED AS ALIORC  
   LIFECYCLE 90;
   
   INSERT OVERWRITE TABLE rec_sln_demo_user_table_preprocess_all_feature_v1 PARTITION (ds)
   SELECT *
   FROM pai_online_project.rec_sln_demo_user_table_preprocess_all_feature_v1
   WHERE ds >= '20231022' and ds <='20231024';

   After the command finishes, the table will contain data from the following partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Item table: rec_sln_demo_item_table_preprocess_all_feature_v1

   CREATE TABLE IF NOT EXISTS rec_sln_demo_item_table_preprocess_all_feature_v1
   like pai_online_project.rec_sln_demo_item_table_preprocess_all_feature_v1
   STORED AS ALIORC  
   LIFECYCLE 90;
   
   INSERT OVERWRITE TABLE rec_sln_demo_item_table_preprocess_all_feature_v1 PARTITION(ds)
   SELECT *
   FROM pai_online_project.rec_sln_demo_item_table_preprocess_all_feature_v1
   WHERE ds >= '20231022' and ds <='20231024';

   After the command finishes, the table will contain data from the following partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Label table: rec_sln_demo_label_table

   CREATE TABLE IF NOT EXISTS rec_sln_demo_label_table
   like pai_online_project.rec_sln_demo_label_table
   STORED AS ALIORC  
   LIFECYCLE 90;
   
   INSERT OVERWRITE TABLE rec_sln_demo_label_table PARTITION (ds)
   SELECT *
   FROM pai_online_project.rec_sln_demo_label_table
   WHERE ds >= '20231022' and ds <='20231024'

   After the command finishes, the table will contain data from the following partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Sequence feature table: rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3

   CREATE TABLE IF NOT EXISTS rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3
   like pai_online_project.rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3
   STORED AS ALIORC  
   LIFECYCLE 90;
   
   INSERT OVERWRITE TABLE rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3 PARTITION(ds)
   SELECT *
   FROM pai_online_project.rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3
   WHERE ds >= '20231022' and ds <='20231024';

   After the command finishes, the table will contain data from the following partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Behavior table: rec_sln_demo_behavior_table_preprocess_v3

   CREATE TABLE IF NOT EXISTS rec_sln_demo_behavior_table_preprocess_v3
   like pai_online_project.rec_sln_demo_behavior_table_preprocess_v3
   STORED AS ALIORC  
   LIFECYCLE 90;
   
   
   INSERT OVERWRITE TABLE rec_sln_demo_behavior_table_preprocess_v3 PARTITION(ds)
   SELECT *
   FROM pai_online_project.rec_sln_demo_behavior_table_preprocess_v3
   WHERE ds >= '20231022' and ds <='20231024';

You can now view the synchronized tables in your workspace. The rest of this tutorial uses these five tables.

Configure data sources

FeatureStore typically requires two types of data sources: an offline data source, such as MaxCompute, and an online data source, such as FeatureDB, Hologres, or Tablestore. This tutorial uses MaxCompute and FeatureDB as examples.

1. Log on to the PAI console. In the navigation pane on the left, choose Data Preparation > FeatureStore.

2. Select a workspace and click Enter FeatureStore.

3. Configure a MaxCompute data source.

   1. On the Create Store tab, click Create Store, and in the page that appears, configure the parameters for the MaxCompute data source.

      Parameter	Recommended value
      Type	MaxCompute
      Name	Enter a custom name.
      MaxCompute Project Name	Select the MaxCompute project that you created.

   2. Click Submit.

4. Configure a FeatureDB data source.

   1. If you have already created a FeatureDB data source, you can skip this step.

   2. On the Store tab, click Create Store and configure the parameters for the FeatureDB data source in the dialog box.

      Parameter	Recommended value
      Type	FeatureDB (If this is your first time using FeatureDB, follow the on-screen instructions to activate it.)
      Name	Custom names are not supported. The default name is feature_db.
      Username	Set a username.
      Password	Set a password.
      VPC network high-speed connection (optional)	This option allows you to use the FeatureStore SDK to directly access FeatureDB from within a VPC over a private network. This connection improves data read/write performance and reduces access latency.
      VPC	Select the VPC that contains your online FeatureStore services.
      Zone and vSwitch	Select a zone and vSwitch. Ensure the vSwitch is in the same zone as your online services. For high availability, we recommend selecting vSwitches in at least two zones.

   3. Click Submit.

FeatureStore project

You can create multiple projects in FeatureStore, with each project serving as an isolated namespace. For detailed instructions, see Configure a FeatureStore project. Running this notebook requires a backend FeatureStore service and configured data sources. After activating FeatureStore, you must configure a data source. For more information, see Create a data source.

The offline_datasource_id parameter specifies the ID of the offline data source, and online_datasource_id specifies the ID of the online data source.

This example uses a project named fs_demo.

access_id = ''
access_ak = ''
region = 'cn-beijing'
fs = FeatureStoreClient(access_key_id=access_id, access_key_secret=access_ak, region=region)
cur_project_name = "fs_demo"
project = fs.get_project(cur_project_name)
if project is None:
  raise ValueError(f"The project '{cur_project_name}' does not exist. You must create it first.")

Run the following code to retrieve the current project and print its information.

project = fs.get_project(cur_project_name)
print(project)

Feature Entity

A feature entity describes a collection of related features. Multiple feature views can be associated with a single feature entity. Each entity has a join_id, which joins features from different feature views. Each feature view has a primary key for retrieving its feature data, and the name of the primary key can be different from the join_id.

In recommendation systems, features are typically associated with either a user or an item entity. This example demonstrates how to create both a user entity and an item entity.

• Create user entity

  user_entity_name = "user"
  user_join_id = 'user_id'
  user_entity = project.get_entity(user_entity_name)
  if user_entity is None:
   user_entity = project.create_entity(name = user_entity_name, join_id=user_join_id)
  user_entity.print_summary()
  

• Create item entity

  item_entity_name = "item"
  join_id = 'item_id'
  item_entity = project.get_entity(item_entity_name)
  if item_entity is None:
   item_entity = project.create_entity(name = item_entity_name, join_id=join_id)
  item_entity.print_summary()

Feature View

A feature view is a core object in FeatureStore that ingests external data. It defines where the data comes from (the data source), any preprocessing or transformation operations, the data structure (the feature schema, including feature names and types), and the storage location (online store and/or offline store). A feature view also manages metadata, such as the primary key, event time, partition key, associated feature entity, and TTL (Time to Live). A default TTL of -1 indicates that the feature is permanently valid. A positive TTL value means that online queries will retrieve only the latest feature data within the specified time window.

Feature views are available in three types: Batch FeatureView (for offline or T-1 day features), Stream FeatureView (for real-time features), and Sequence FeatureView (for sequence features).

Batch FeatureView

A Batch FeatureView injects offline data into the FeatureStore offline store and can optionally synchronize it to the online store to support real-time queries. Use this feature view for offline or T-1 day features.

• Register the user offline feature table

  1. Register the rec_sln_demo_user_table_preprocess_all_feature_v1 table in FeatureStore.

     user_feature_view_name = "user_table_preprocess_all_feature_v1"
     user_table_name = "rec_sln_demo_user_table_preprocess_all_feature_v1"
     user_feature_view = project.get_feature_view(user_feature_view_name)
     if user_feature_view is None:
      ds = MaxComputeDataSource(project.offline_datasource_id, user_table_name)
      user_feature_view = project.create_batch_feature_view(name=user_feature_view_name, datasource=ds, online=True, entity= user_entity_name, primary_key='user_id', register=True)
     print(user_feature_view)

  2. Synchronize data in the 20231023 partition of the rec_sln_demo_user_table_preprocess_all_feature_v1 table from the offline data source to the online data source.

     user_task = user_feature_view.publish_table({'ds':'20231023'})
     user_task.wait()

  3. Check the task status.

     user_task.print_summary()

• Register the item offline feature table

  1. Register the rec_sln_demo_item_table_preprocess_all_feature_v1 table in FeatureStore.

     item_feature_view_name = "item_table_preprocess_all_feature_v1"
     item_table_name = "rec_sln_demo_item_table_preprocess_all_feature_v1"
     item_feature_view = project.get_feature_view(item_feature_view_name)
     if item_feature_view is None:
       ds = MaxComputeDataSource(project.offline_datasource_id, item_table_name)
       item_feature_view = project.create_batch_feature_view(name=item_feature_view_name, datasource=ds, online = True, entity= item_entity_name, primary_key='item_id', register=True)
     print(item_feature_view)

  2. Synchronize data in the 20231023 partition of the rec_sln_demo_item_table_preprocess_all_feature_v1 table from the offline data source to the online data source.

     item_task = item_feature_view.publish_table({'ds':'20231023'})
     item_task.wait()

  3. Check the task status.

     item_task.print_summary()

Sequence FeatureView

A Sequence FeatureView supports writing sequence features offline and reading real-time sequence features online. In recommendation scenarios, an offline sequence feature table (F1) is initially generated through simulation and can later be replaced by online logs. During online queries for real-time sequences, data is retrieved from two online behavior tables: a T-1 day behavior table (B1) and a current-day (T-day) real-time behavior table (B2). The B2 table contains features updated by real-time computation. FeatureStore queries both B1 and B2, constructs the user's feature sequence, and then combines it with other features for model scoring.

The online T-1 day behavior table (B1) is typically synchronized from an offline T-1 day behavior table (A1), with FeatureStore automatically handling operations like deduplication. The current-day (T-day) online behavior table (B2) currently requires you to write data through an API or other Alibaba Cloud products such as Flink.

Therefore, when you register a Sequence FeatureView, FeatureStore manages four tables on your behalf: the offline sequence table (F1), the offline T-1 day behavior table (A1), the online T-1 day behavior table (B1), and the online T-day behavior table (B2).

During registration, you only provide the offline sequence table (F1) and the offline T-1 day behavior table (A1). FeatureStore creates the online tables and handles synchronization and deduplication.

1. Register the Sequence FeatureView.

   seq_feature_view_name = "wide_seq_feature_v3"
   seq_feature_view = project.get_feature_view(seq_feature_view_name)
   if seq_feature_view is None:
     seq_table_name = "rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3"
     behavior_table_name = 'rec_sln_demo_behavior_table_preprocess_v3'
     ds = MaxComputeDataSource(project.offline_datasource_id, behavior_table_name)
     event_time = 'event_unix_time' # Field name for event time in the behavior table.
     item_id = 'item_id' # Field name for item_id in the behavior table.
     event = 'event' # Field name for the event in the behavior table.
     # deduplication_method = 1: Deduplicates by ['user_id', 'item_id', 'event'].
     # deduplication_method = 2: Deduplicates by ['user_id', 'item_id', 'event', 'event_time'].
     sequence_feature_config_list = [SequenceFeatureConfig(offline_seq_name='click_seq_50_seq', seq_event='click', online_seq_name='click_seq_50', seq_len=50)]
     # offline_seq_name: Column name of the sequence feature in the offline sequence table.
     # seq_event: The behavior field name used for filtering.
     # online_seq_name: The name under which the corresponding item_id sequence for the user is returned by the online Go SDK.
     # seq_len: The sequence length. Sequences longer than this value are truncated.
     seq_table_config = SequenceTableConfig(table_name=seq_table_name, primary_key='user_id', event_time='event_unix_time')
     seq_feature_view = project.create_sequence_feature_view(seq_feature_view_name, datasource=ds,
                                                 event_time=event_time, item_id=item_id, event=event, deduplication_method=1,
                                                 sequence_feature_config=sequence_feature_config_list, sequence_table_config=seq_table_config, entity=user_entity_name)
   # seq_feature_view.print_summary()
   print(seq_feature_view)

2. Synchronize data in the 20231023 partition of the rec_sln_demo_behavior_table_preprocess_v3 table from the offline data source to the online data source. During synchronization, the system automatically checks for data from the previous N days and backfills it if missing. You can specify N by using the days_to_load parameter, which defaults to 30. We recommend using the default value in most cases.

   seq_task = seq_feature_view.publish_table({'ds':'20231023'}, days_to_load=30)
   seq_task.wait()

3. Check the task status.

   seq_task.print_summary()

Stream FeatureView

A Stream FeatureView writes data directly to the online store and simultaneously syncs it to the offline store. Use it for scenarios that require real-time feature updates, such as updating item prices or sales volumes.

Register the label table

label_table_name = 'rec_sln_demo_label_table'
ds = MaxComputeDataSource(data_source_id=project.offline_datasource_id, table=label_table_name)
label_table = project.get_label_table(label_table_name)
if label_table is None:
  label_table = project.create_label_table(datasource=ds, event_time='event_unix_time')
print(label_table)

Retrieve online features

You can retrieve features from the online store for purposes such as debugging offline-online consistency and data analysis. This feature is currently optimized for Hologres.

user_feature_view_name = "user_table_preprocess_all_feature_v1"
user_feature_view = project.get_feature_view(user_feature_view_name)
ret_features_1 = user_feature_view.list_feature_view_online_features(join_ids=['169898460', '148811946'])
print("ret_features = ", ret_features_1)

Training set

To train a model, first construct a training set of label and feature data. When interacting with FeatureStore, you provide the label data, specify the features to retrieve, and then the system performs a point-in-time join based on the primary key (if an event time exists).

# Specify the label table
label_table_name = 'rec_sln_demo_label_table'

output_ds = MaxComputeDataSource(data_source_id=project.offline_datasource_id)
train_set_output = TrainingSetOutput(output_ds)

user_feature_view_name = "user_table_preprocess_all_feature_v1"
user_feature_selector = FeatureSelector(user_feature_view_name, '*') # '*' selects all features.
item_feature_view_name = "item_table_preprocess_all_feature_v1"
item_feature_selector = FeatureSelector(item_feature_view_name, '*')
seq_feature_view_name = "wide_seq_feature_v3"
seq_feature_selector = FeatureSelector(seq_feature_view_name, ['click_seq_50_seq'])
train_set = project.create_training_set(label_table_name=label_table_name, train_set_output= train_set_output, feature_selectors=[user_feature_selector, item_feature_selector, seq_feature_selector])
print("train_set = ", train_set)

Model features

After training a model and deploying it as a service, you can use it for business predictions. The training samples are available in the train_set object from the previous step.

model_name = "fs_rank_v2"
cur_model = project.get_model(model_name)
if cur_model is None:
  cur_model = project.create_model(model_name, train_set)
print("cur_model_train_set_table_name = ", cur_model.train_set_table_name)