Platform For AI:Use FeatureStore SDK for Python to build a recommendation system

Synchronize data from simulated tables

A recommendation system requires the following tables: user feature table, item feature table, label table, sequence feature table, and behavior table.

This topic uses the simulated tables in the MaxCompute project pai_online_project as examples to show how to use FeatureStore SDK for Python. You must execute SQL statements in DataWorks to synchronize the simulated tables from the pai_online_project project to your MaxCompute project. To synchronize the simulated tables, perform the following steps:

1. Log on to the DataWorks console.

2. In the left-side navigation pane, choose Data Development and Governance > DataStudio.

3. On the DataStudio page, select the DataWorks workspace that you created and click Go to DataStudio.

4. Move the pointer over the Create button and choose Create Node > MaxCompute > ODPS SQL. In the Create Node dialog box, configure the node parameters described in the following table.

   Parameter	Description
   Engine Instance	Select the MaxCompute engine that you created.
   Node Type	Select ODPS SQL from the Node Type drop-down list.
   Path	Choose Business Flow > Workflow > MaxCompute.
   Name	Specify a name for the node.

5. Click Confirm.

6. On the tab of the node that you created, run the following SQL statements to synchronize the simulated user table, item table, label table, sequence table, and behavior table from the pai_online_project project to your MaxCompute project. Select the exclusive resource group that you created as the resource group.

   Synchronize the user table rec_sln_demo_user_table_preprocess_all_feature_v1 (Click to view 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'

   The data in the following partitions is synchronized:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the item table rec_sln_demo_item_table_preprocess_all_feature_v1 (Click to view 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'

   The data in the following partitions is synchronized:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the label table rec_sln_demo_label_table (Click to view 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'

   The data in the following partitions is synchronized:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the sequence table rec_sln_demo_behavior_table_preprocess_sequence_wide_seq_feature_v3 (Click to view 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'

   The data in the following partitions is synchronized:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Synchronize the behavior table rec_sln_demo_behavior_table_preprocess_v3 (Click to view 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 perform the preceding operations, you can view the synchronized tables in your workspace. These tables are used as examples in the following sections.

Configure data stores

In most cases, you need to configure an offline data store, such as a MaxCompute project, and an online data store, such as a Hologres instance, a GraphCompute instance, or a Tablestore instance, in FeatureStore. In this example, a MaxCompute project is configured as an offline data store, and a Hologres instance is configured as an online data store.

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

2. On the FeatureStore page, select a workspace from the drop-down list and click Enter FeatureStore.

3. Configure a MaxCompute data store.

   1. On the Store tab, click Create Store. In the Create Store panel, configure the parameters described in the following table.

      Parameter	Description
      Type	Select MaxCompute from the drop-down list.
      Name	Specify a name for the data store.
      MaxCompute Project Name	Select the MaxCompute project that you created.

   2. Copy the authorization statement and click Go to to execute the copied statement in DataWorks. In this way, the Hologres instance is authorized to synchronize data from the MaxCompute project.

      Note

      To grant permissions to the Hologres instance, make sure that your account has admin permissions. For more information, see Manage user permissions by using commands or Manage user permissions in the MaxCompute console (new version).

   3. Click Submit.

4. Configure a Hologres data store.

   1. On the Store tab, click Create Store. In the Create Store panel, configure the parameters described in the following table.

      Parameter	Description
      Type	Select Hologres from the drop-down list.
      Name	Specify a name for the data store.
      Instance ID	Select the Hologres instance that you created.
      Database Name	Select the database that you created in the Hologres instance.

   2. Click Submit.

   3. Grant the permissions to access the Hologres instance. For more information, see Configure data stores.

FeatureStore project

You can create multiple projects in FeatureStore. Each project is independent. For more information, see Configure FeatureStore projects. To run the sample code in Jupyter Notebook, specific configurations are required in FeatureStore. For example, you must configure data stores in FeatureStore. For more information, see Configure data stores.

In the sample code in this topic, the offline_datasource_id parameter specifies the ID of the offline data store, and the online_datasource_id parameter specifies the ID of the online data store.

The following sample code uses the fs_movie project as an example to show how to configure a FeatureStore project.

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")

Obtain the project and print the project information.

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

Feature entity

A feature entity is a collection of semantically related features. A feature entity can be associated with multiple feature views. Each entity has a join ID. You can use join IDs to associate features across multiple feature views. Each feature view has a primary key (index) that can be used to retrieve features. The primary key can be different from the name of the join ID.

Typically, a recommendation system has two feature entities: user and item. The following sample code provides an example on how to create user and item entities:

• 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

You can create feature views to ingest feature data from external sources into FeatureStore for centralized management. A feature view contains all information required to manage features, including the data source, required transformations, feature table schema, and online and offline data stores. A feature view also allows you to manage the metadata, including primary keys, event time, partition fields, feature entities, and time to live (TTL). TTL is a parameter that specifies the period of time during which feature data is available in the online data store. The default value of -1 specifies that the online data store retains all feature data. A positive value specifies that the online data store retains only the most recent feature data within the specified time period.

FeatureStore supports the following types of feature views: batch feature view, stream feature view, and sequence feature view.

Batch feature view

A batch feature view allows you to ingest offline features into the offline data store and synchronize offline features to the online data store. This way, you can query features in real time.

• Register the user table.

  1. Register the user table rec_sln_demo_user_table_preprocess_all_feature_v1 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 ds=20231023 partition from the offline data store to the online data 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 table.

  1. Register the item table rec_sln_demo_item_table_preprocess_all_feature_v1 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 ds=20231023 partition from the offline data store to the online data store.

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

  3. View the task running status.

     item_task.print_summary()

Sequence feature view

A sequence feature view allows you to write sequence features offline and read real-time sequence features online. When you build a recommendation system, you can register a simulated offline sequence table (F1 table) in FeatureStore. The data in the simulated table is subsequently replaced by online logs. To query real-time sequence features, the system retrieves data from the online behavior table of the current day (B2 table) and the online behavior table of the previous day (B1 table). The data retrieved from these two tables is concatenated to generate sequence features, which are then sent to a model for scoring.

The B1 table synchronizes data from the offline behavior table of the previous day (A1 table). The duplicated data is automatically filtered by FeatureStore during synchronization. The B2 table contains the real-time behavior features. You can call API operations or use Realtime Compute for Apache Flink to write real-time data to the B2 table.

After you create a sequence feature view, FeatureStore manages the F1, A1, B1, and B2 tables in a centralized manner.

You need to register only the F1 and A1 tables in the offline data store to the sequence feature view. FeatureStore automatically creates the online behavior tables (B1 and B2 tables) by synchronization and deduplication.

1. Create a sequence 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 specifies to deduplicate data based on the user_id, item_id, and event fields.
     # deduplication_method = 2 specifies to deduplicate data based on the user_id, item_id, event, and event_time fields.
     sequence_feature_config_list = [SequenceFeatureConfig(offline_seq_name='click_seq_50_seq', seq_event='click', online_seq_name='click_seq_50', seq_len=50)]
     # The offline_seq_name parameter specifies the name of the sequence feature field in the offline sequence table. The seq_event parameter specifies the name of the event field. The online_seq_name parameter specifies the field name for the online sequence features retrieved by using FeatureStore SDK for Go. 
     # The seq_len parameter specifies the maximum length of the sequence feature. Sequences that exceed the limit 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 ds=20231023 partition of the behavior table rec_sln_demo_behavior_table_preprocess_v3 from the offline data store to the online data store. During synchronization, FeatureStore automatically checks and loads historical data if the data is missing. The days_to_load parameter specifies the number of previous days for which FeatureStore checks the data availability. The default value of this parameter is 30.

   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 feature view

A stream feature view allows you to write real-time features to the online data store and synchronize data to the offline data store. This works for scenarios where features are updated in real time, such as the price and sales of goods.

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 online features to analyze data and check whether features are consistent in the online and offline data stores.

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)

Training dataset

You can use FeatureStore to generate a training dataset for model training. A training dataset contains labels and features. You must prepare the labels for model training and define the features that the model needs to fetch from the feature views. The labels are associated with the features by using point-in-time joins based on the primary keys.

# 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, '*') # '*' Specifies 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

You can train a model on the training dataset generated by FeatureStore and deploy the trained model as an inference service in PAI.

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)