Platform For AI:Manage features in a recommendation system by using FeatureStore

Sync data tables

A typical recommendation scenario requires three data tables: user features, item features, and labels. This tutorial uses simulated tables in the pai_online_project MaxCompute project.

Run the following SQL in DataWorks to sync these tables from pai_online_project to your MaxCompute project.

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='${bdp.system.bizdate}')
SELECT * except(ds)
FROM pai_online_project.rec_sln_demo_user_table_preprocess_all_feature_v1
WHERE ds = '${bdp.system.bizdate}';

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='${bdp.system.bizdate}')
SELECT * except(ds)
FROM pai_online_project.rec_sln_demo_item_table_preprocess_all_feature_v1
WHERE ds = '${bdp.system.bizdate}';

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='${bdp.system.bizdate}')
SELECT * except(ds)
FROM pai_online_project.rec_sln_demo_label_table
WHERE ds = '${bdp.system.bizdate}';

Configure data sources

FeatureStore requires an offline store (MaxCompute) and an online store (FeatureDB, Hologres, or Tablestore). This tutorial uses MaxCompute and FeatureDB.

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

2. Select a workspace and click Enter FeatureStore.

3. Configure the MaxCompute data source. On the Data Source tab, click Create Store. Set the following parameters.

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

4. Configure the FeatureDB data source. If you already created one, skip this step. On the Store tab, click Create Store. Set the following parameters. Click Submit.

   Parameter	Suggested value
   Type	FeatureDB (If first use, follow on-screen prompts to activate)
   Name	Cannot be customized. Default value is feature_db.
   Username	Set a username.
   Password	Set a password.
   VPC high-speed connection (Optional)	After successful configuration, use the FeatureStore SDK in a VPC to access FeatureDB through PrivateLink. This reduces latency and improves performance.
   VPC	Select the VPC where your online FeatureStore service is located.
   Zone and vSwitch	Select a zone and vSwitch. Select the vSwitch for the zone where your online service machine is located. Select vSwitches in at least two zones for high availability.

Install the FeatureStore Python SDK

Install the SDK on the DataWorks exclusive resource group for scheduling.

1. Log on to the DataWorks console.

2. In the navigation pane on the left, click Resource Group.

3. On the Exclusive Resource Groups tab, find the resource group where Purpose is set to Data Scheduling. Click the icon for the schedule resource and choose O&M Assistant.

4. Click Create Command. Set the following parameters.

   Parameter	Suggested value
   Command Name	Enter a custom name. This tutorial uses install.
   Command Type	Manual Input (pip Command Cannot Be Used To Install Third-party Packages)
   Command Content	/home/tops/bin/pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple https://feature-store-py.oss-cn-beijing.aliyuncs.com/package/feature_store_py-2.0.2-py3-none-any.whl
   Timeout	Enter a custom time.

5. Click OK to create the command.

6. Click Command. In the page that appears, click Running.

7. Click the button to view the latest execution status. When the status changes to Successful, the installation is complete.

Method 1: Use the console

Method 2: Use Python SDK

For SDK usage steps, see DSW Gallery.

Export the training set

1. Log on to the DataWorks console.

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

3. Select the DataWorks workspace that you created and click Go to Data Studio.

4. Hover over Create, and choose Create Node > MaxCompute > PyODPS 3. Set the following parameters.

   Parameter	Suggested value
   Engine Instance	Select the MaxCompute engine you created.
   Node Type	PyODPS 3
   Path	Business Flow/Workflow/MaxCompute
   Name	Enter a custom name.

5. Click Confirm.

6. Copy the following script into the node.

from feature_store_py.fs_client import FeatureStoreClient
from feature_store_py.fs_project import FeatureStoreProject
from feature_store_py.fs_datasource import LabelInput, MaxComputeDataSource, TrainingSetOutput
from feature_store_py.fs_features import FeatureSelector
from feature_store_py.fs_config import LabelInputConfig, PartitionConfig, FeatureViewConfig
from feature_store_py.fs_config import TrainSetOutputConfig, EASDeployConfig
import datetime
import sys
from odps.accounts import StsAccount

cur_day = args['dt']
print('cur_day = ', cur_day)
offset = datetime.timedelta(days=-1)
pre_day = (datetime.datetime.strptime(cur_day, "%Y%m%d") + offset).strftime('%Y%m%d')
print('pre_day = ', pre_day)


access_key_id = o.account.access_id
access_key_secret = o.account.secret_access_key
sts_token = None
endpoint = 'paifeaturestore-vpc.cn-beijing.aliyuncs.com'
if isinstance(o.account, StsAccount):
    sts_token = o.account.sts_token
fs = FeatureStoreClient(access_key_id=access_key_id, access_key_secret=access_key_secret, security_token=sts_token, endpoint=endpoint)
cur_project_name = 'fs_demo'
project = fs.get_project(cur_project_name)


label_partitions = PartitionConfig(name = 'ds', value = cur_day)
label_input_config = LabelInputConfig(partition_config=label_partitions)

user_partitions = PartitionConfig(name = 'ds', value = pre_day)
feature_view_user_config = FeatureViewConfig(name = 'user_table_preprocess_all_feature_v1',
partition_config=user_partitions)

item_partitions = PartitionConfig(name = 'ds', value = pre_day)
feature_view_item_config = FeatureViewConfig(name = 'item_table_preprocess_all_feature_v1',
partition_config=item_partitions)
feature_view_config_list = [feature_view_user_config, feature_view_item_config]
train_set_partitions = PartitionConfig(name = 'ds', value = cur_day)
train_set_output_config = TrainSetOutputConfig(partition_config=train_set_partitions)


model_name = 'fs_rank_v1'
cur_model = project.get_model(model_name)
task = cur_model.export_train_set(label_input_config, feature_view_config_list, train_set_output_config)
task.wait()
print("task_summary = ", task.task_summary)

1. In the right-side pane, click Scheduling Configuration. Set the following parameters.

   Parameter	Suggested value
   Scheduling Parameters
   Parameter Name	dt
   Parameter Value	$[yyyymmdd-1]
   Resource Properties
   Scheduling Resource Group	Select the exclusive resource group for scheduling that you created.
   Scheduling Dependencies	Select the user table and item table you created.

2. After configuring and testing the node, save and submit the node configuration.

3. Perform data backfill.

View export task (optional)

1. On the FeatureStore Projects page, click a project name to open its details page.

2. On the Feature Entity tab, click Jobs.

3. Click Details in the row of the target task to view its basic information, run configuration, and task logs.

Train the model

EasyRec is an open-source recommendation framework that integrates with FeatureStore to train, export, and publish models. Use the fs_demo_fs_rank_v1_trainning_set table as input to train a model with EasyRec.

For EasyRec questions, join PAI consultation group on DingTalk (Group ID: 32260796).

Schedule data synchronization

Schedule regular synchronization of user and item feature tables from the offline store to the online store.

1. Log on to the DataWorks console.

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

3. Select the DataWorks workspace that you created and click Go to Data Studio.

4. Schedule synchronization for user_table_preprocess_all_feature_v1 (Click to view details)

   1. Hover over New, and choose New Node > MaxCompute > PyODPS 3.

   2. Copy the following script to schedule synchronization for user_table_preprocess_all_feature_v1. ``python from feature_store_py.fs_client import FeatureStoreClient import datetime from feature_store_py.fs_datasource import MaxComputeDataSource import sys cur_day = args['dt'] print('cur_day = ', cur_day) access_key_id = o.account.access_id access_key_secret = o.account.secret_access_key fs = FeatureStoreClient(access_key_id=access_key_id, access_key_secret=access_key_secret, region='cn-beijing') cur_project_name = 'fs_demo' project = fs.get_project(cur_project_name) feature_view_name = 'user_table_preprocess_all_feature_v1' batch_feature_view = project.get_feature_view(feature_view_name) task = batch_feature_view.publish_table(partitions={'ds':cur_day}, mode='Overwrite') task.wait() task.print_summary() ``

   3. In the right-side pane, click Scheduling Configuration. Set the following parameters. | Parameter | Suggested value | |---|---| | Scheduling Parameters | | | Parameter Name | dt | | Parameter Value | $[yyyymmdd-1] | | Resource Properties | | | Scheduling Resource Group | Select the exclusive resource group for scheduling that you created. | | Scheduling Dependencies | Select the user table that you created. |

   4. After configuring and testing the node, save and submit the node configuration.

   5. Perform data backfill.

5. Schedule synchronization for item_table_preprocess_all_feature_v1 (Click to view details)

   1. Hover over Create, and choose Create Node > MaxCompute > PyODPS 3. Configure the node parameters.

   2. Click Confirm.

   3. Copy the following script. ``python from feature_store_py.fs_client import FeatureStoreClient import datetime from feature_store_py.fs_datasource import MaxComputeDataSource import sys cur_day = args['dt'] print('cur_day = ', cur_day) access_key_id = o.account.access_id access_key_secret = o.account.secret_access_key fs = FeatureStoreClient(access_key_id=access_key_id, access_key_secret=access_key_secret, region='cn-beijing') cur_project_name = 'fs_demo' project = fs.get_project(cur_project_name) feature_view_name = 'item_table_preprocess_all_feature_v1' batch_feature_view = project.get_feature_view(feature_view_name) task = batch_feature_view.publish_table(partitions={'ds':cur_day}, mode='Overwrite') task.wait() task.print_summary() ``

   4. In the right-side pane, click Scheduling Configuration. Set the following parameters. | Parameter | Suggested value | |---|---| | Scheduling Parameters | | | Parameter Name | dt | | Parameter Value | $[yyyymmdd-1] | | Resource Properties | | | Scheduling Resource Group | Select the exclusive resource group for scheduling that you created. | | Scheduling Dependencies | Select the item table that you created. |

   5. After configuring and testing the node, save and submit the node configuration.

   6. Perform data backfill.

Create and deploy EAS model service

EasyRec Processor includes the FeatureStore C++ SDK for low-latency, high-performance feature retrieval. It receives scoring requests from the recommendation engine, fetches features, runs inference, and returns scores.

1. Log on to the DataWorks console.

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

3. Select the DataWorks workspace that you created and click Go to Data Studio.

4. Hover over New, and choose New Node > MaxCompute > PyODPS 3.

5. Copy the following script.

import os
import json
config = {
  "name": "fs_demo_v1",
  "metadata": {
    "cpu": 4,
    "rpc.max_queue_size": 256,
    "rpc.enable_jemalloc": 1,
    "gateway": "default",
    "memory": 16000
  },
  "model_path": f"oss://beijing0009/EasyRec/deploy/rec_sln_demo_dbmtl_v1/{args['ymd']}/export/final_with_fg",  # Path of the trained model. customize the path.
  "model_config": {
    "access_key_id": f'{o.account.access_id}',
    "access_key_secret": f'{o.account.secret_access_key}',
    "region": "cn-beijing",  # Replace this with the region where PAI is deployed. This tutorial uses cn-beijing as an example.
    "fs_project": "fs_demo",  # Replace this with the name of your FeatureStore project. This tutorial uses fs_demo as an example.
    "fs_model": "fs_rank_v1",  # Replace this with the name of your FeatureStore model feature. This tutorial uses fs_rank_v1 as an example.
    "fs_entity": "item",
    "load_feature_from_offlinestore": True,
    "steady_mode": True,
    "period": 2880,
    "outputs": "probs_is_click,y_ln_playtime,probs_is_praise",
    "fg_mode": "tf"
  },
  "processor": "easyrec-1.9",
  "processor_type": "cpp"
}

with open("echo.json", "w") as output_file:
    json.dump(config, output_file)

# Run the following line for the first deployment
os.system(f"/home/admin/usertools/tools/eascmd -i {o.account.access_id} -k {o.account.secret_access_key} -e pai-eas.cn-beijing.aliyuncs.com create echo.json")

# Run the following line for scheduled updates
# os.system(f"/home/admin/usertools/tools/eascmd -i {o.account.access_id} -k {o.account.secret_access_key} -e pai-eas.cn-beijing.aliyuncs.com modify fs_demo_v1 -s echo.json")

1. In the right-side pane, click Scheduling Configuration. Set the following parameters.

   Parameter	Suggested value
   Scheduling Parameters
   Parameter Name	dt
   Parameter Value	$[yyyymmdd-1]
   Resource Properties
   Scheduling Resource Group	Select the exclusive resource group for scheduling you created.
   Scheduling Dependencies	Select the corresponding training task and item_table_preprocess_all_feature_v1.

2. After configuring and testing the node, run it and check deployment status.

3. After deployment completes, comment out line 34, uncomment line 37, and submit the task for scheduled execution. > Tip: Line 34 uses create for first-time deployment. Line 37 uses modify for subsequent scheduled updates. After initial deployment succeeds, switch to the modify command.

4. (Optional) View the deployed service on the Inference Service tab of the Elastic Algorithm Service (EAS) page. Custom deployment.

5. (Optional) If your data source requires VPC access (such as Hologres), connect the EAS and data source VPCs. For Hologres:

   1. Find VPC ID and vSwitch ID on the Network Information page of the Hologres instance.

   2. On the EAS service page, click Configure High-Speed Connection in the upper-right corner.

   3. Enter the corresponding VPC ID and vSwitch ID.

   4. Enter a Security Group Name. Select an existing security group or create one. Hologres typically uses port 80. The security group must allow traffic on this port.

   5. Click OK. The service becomes available after the update completes.

Configure PAI-REC

PAI-REC is a recommendation engine service that integrates the FeatureStore Go SDK. It connects with FeatureStore and EAS.

"FeatureStoreConfs": {
        "pairec-fs": {
            "RegionId": "cn-beijing",
            "AccessId": "${AccessKey}",
            "AccessKey": "${AccessSecret}",
            "ProjectName": "fs_demo"
        }
    },

"FeatureConfs": {
        "recreation_rec": {
            "AsynLoadFeature": true,
            "FeatureLoadConfs": [
                {
                    "FeatureDaoConf": {
                        "AdapterType": "featurestore",
                        "FeatureStoreName": "pairec-fs",
                        "FeatureKey": "user:uid",
                        "FeatureStoreModelName": "fs_rank_v1",
                        "FeatureStoreEntityName": "user",
                        "FeatureStore": "user"
                    }
                }
            ]
        }
    },

"AlgoConfs": [
        {
            "Name": "fs_demo_v1",
            "Type": "EAS",
            "EasConf": {
                "Processor": "EasyRec",
                "Timeout": 300,
                "ResponseFuncName": "easyrecMutValResponseFunc",
                "Url": "eas_url_xxx",
                "EndpointType": "DIRECT",
                "Auth": "eas_token"
            }
        }
    ],