Platform For AI:Build a recommendation system using the FeatureStore Python SDK

Synchronize data tables

For a recommendation scenario, you generally need to prepare the following data tables: a user feature table, an item feature table, a label table, a sequence feature table, and a behavior table.

To simplify this tutorial, we have prepared sample user, item, label, sequence feature, and behavior tables in the pai_online_project MaxCompute project. You need to run SQL commands in DataWorks to synchronize these tables from the pai_online_project project to your own MaxCompute project. The procedure is as follows:

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 choose New Node > MaxCompute > ODPS SQL. In the dialog box that appears, configure the node parameters.

   Parameter	Suggested value
   Engine Instance	Select your MaxCompute engine.
   Node Type	ODPS SQL
   Path	Business Flow/Workflow/MaxCompute
   Name	You can specify a custom name.

5. Click Confirm.

6. In the node editor, run the following SQL commands to synchronize the user, item, label, sequence feature, and behavior tables from the pai_online_project project to your MaxCompute project. For the resource group, select the exclusive resource group that you created.

   Synchronize the user table: rec_sln_demo_user_table_preprocess_all_feature_v1 (Click for details)

   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 you run the command, the command returns data from the following three partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the item table: rec_sln_demo_item_table_preprocess_all_feature_v1 (Click for details)

   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 you run the command, the command returns data from the following three partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the label table: rec_sln_demo_label_table (Click for details)

   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 you run the command, the command returns data from the following three partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the sequence feature table: rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3 (Click for details)

   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 you run the command, the command returns data from the following three partitions:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the behavior table: rec_sln_demo_behavior_table_preprocess_v3 (Click for details)

   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';

After you complete these steps, the synchronized tables are available in your workspace. These five tables are used as examples in the following sections.

Configure data sources

You typically need to configure two data sources for FeatureStore: an offline data source (MaxCompute) and an online data source (FeatureDB, Hologres, or TableStore). This topic uses MaxCompute and FeatureDB as examples.

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

2. Select a workspace and click Enter FeatureStore.

3. Configure a MaxCompute data source.

   1. On the Data Source tab, click Create Store. In the dialog box that appears, configure the MaxCompute data source parameters.

      Parameter	Suggested value
      Type	MaxCompute
      Name	Enter a custom name.
      MaxCompute Project Name	Select your MaxCompute project.

   2. After the configuration is complete, 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. In the dialog box that appears, configure the FeatureDB data source parameters.

      Parameter	Suggested value
      Type	FeatureDB (If this is your first time using it, follow the on-screen instructions to activate FeatureDB)
      Name	Custom names are not supported. The default value is feature_db.
      Username	Set a username.
      Password	Set a password.
      VPC High-speed Connection (Optional)	After the configuration is successful, you can use the FeatureStore SDK in a VPC to directly access FeatureDB through a PrivateLink connection. This improves data read and write performance and reduces access latency.
      VPC	Select the VPC where your online FeatureStore service is located.
      Zone and vSwitch	Select a zone and vSwitch. Make sure to select the vSwitches in the zones where your online service machines are located. We recommend that you select vSwitches in at least two zones to ensure high availability and stability for your services.

   3. After the configuration is complete, click Submit.

Feature engineering project

You can create multiple independent projects in FeatureStore. For more information, see Configure a FeatureStore project. To run a Notebook, you need a FeatureStore server-side environment. After you activate FeatureStore, you must also configure data sources. For more information, see Create a new data source.

The `offline_datasource_id` and `online_datasource_id` parameters specify the offline and online data source IDs, respectively.

This topic uses a project named fs_movie as an example.

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("Need to create project : fs_movie")

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

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

Feature entity (FeatureEntity)

A feature entity describes a set of related features. A single feature entity can be associated with multiple feature views. Each entity has a JoinId that is used to associate features from these views. Each feature view has a primary key (index key) to retrieve its feature data, but the name of the index key can be different from the name defined by the JoinId.

In a recommendation system, features are typically associated with only two feature entities: user and item. This means features belong to either the user or the item. This topic provides an example of creating user and item feature entities.

• Create a 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 an 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 (FeatureView)

FeatureStore is a platform for managing and organizing feature data. External data is ingested into FeatureStore through a feature view. A feature view defines the data source, required pre-processing or transformation operations (such as feature engineering or transformation), the data structure of the features (including feature names and types), and the storage location (online store or offline store). It also manages feature metadata, such as primary keys, event times, partition keys, feature entities, and time-to-live (TTL) settings. A TTL of -1 (the default) indicates that the data is stored permanently. A positive number means that online queries will retrieve the latest feature data within the specified TTL.

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

Batch FeatureView (offline feature view)

A Batch FeatureView injects offline data into the offline store of FeatureStore and can synchronize the data to the online store to support real-time queries. This type of view is typically used for offline features or T-1 day features.

• Register the user-side offline feature table

  1. Register the rec_sln_demo_user_table_preprocess_all_feature_v1 table with 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 from the 20231023 partition of the rec_sln_demo_user_table_preprocess_all_feature_v1 table in the offline store to the online store.

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

  3. View the task running status.

     user_task.print_summary()

• Register the item-side offline feature table

  1. Register the rec_sln_demo_item_table_preprocess_all_feature_v1 table with 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 from the 20231023 partition of the rec_sln_demo_item_table_preprocess_all_feature_v1 table in the offline store to the online store.

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

  3. View the task running status.

     item_task.print_summary()

Sequence FeatureView (real-time sequence view)

A Sequence FeatureView supports writing offline sequence features and querying real-time sequence features. In a typical recommendation scenario, the offline sequence feature table (F1) is initially generated from simulated data and can later be replaced by online logs. During an online real-time sequence query, data is queried from two online behavior tables: the T-1 day behavior table (B1) and the real-time behavior table for day T (B2). B2 contains features that are updated in real-time. After data is queried from the B1 and B2 tables, a feature sequence is constructed for the user. This sequence is then combined with other features and sent to the model for scoring.

The online T-1 day behavior table (B1) is typically synchronized from the offline T-1 day behavior table (A1). During synchronization, FeatureStore automatically performs operations such as deduplication. You must write data to the online day T behavior table (B2) using API operations or other Alibaba Cloud products, such as Flink.

Therefore, when you register a real-time feature view, FeatureStore simultaneously manages four tables: the offline sequence table (F1), the offline T-1 day behavior table (A1), the online T-1 day behavior table (B1), and the online day T behavior table (B2).

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

1. Register a real-time feature view.

   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' # The name of the event time field in the behavior table.
     item_id = 'item_id' # The name of the item_id field in the behavior table.
     event = 'event' # The name of the event field in the behavior table.
     # deduplication_method = 1 indicates deduplication based on ['user_id', 'item_id', 'event'].
     # deduplication_method = 2 indicates deduplication based on ['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 is the name of the sequence feature field in the offline sequence table. seq_event is the name of the behavior field. 
     # online_seq_name is the name used for the sequence of item_ids for the user, which is retrieved by the FeatureStore online Go SDK.
     # seq_len is the sequence length. Sequences longer than this length 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 from the 20231023 partition of the rec_sln_demo_behavior_table_preprocess_v3 table in the offline store to the online store. During synchronization, the system automatically checks for data from the previous N days' partitions. If data is missing, it is automatically backfilled. You can specify N using days_to_load. The default value is 30, which is sufficient for most scenarios.

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

3. View the task running status.

   seq_task.print_summary()

Stream FeatureView (real-time feature view)

Writing data directly to the OnlineStore and simultaneously syncing it to the OfflineStore is ideal for scenarios that require real-time feature updates, such as updating product prices and 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)

Get online features

You can retrieve online features to troubleshoot data consistency between the offline and online stores, perform data analytics, and for other purposes. Hologres is the recommended data source.

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.get_online_features(join_ids={'user_id':['169898460', '148811946']}, features=['user_id', 'gender', 'age', 'city'])
print("ret_features = ", ret_features_1)

TrainingSet

When you train a model, you must first construct a sample table. The sample table consists of label data and feature data. To interact with FeatureStore, you must provide the label data and define the names of the features that you want to retrieve. A point-in-time join is then performed based on the primary key and event time, 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 you train a model and deploy it as a service, you can use it for business prediction. The training samples can be obtained from the train_set mentioned previously.

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)