使用FeatureStore Python SDK搭建推荐系统

同步数据表

一般对于推荐场景，需要准备以下数据表：user侧的特征表、item侧的特征表、Label表、序列特征表和行为表。

为方便实践操作，我们在MaxCompute的pai_online_project项目中提前准备了模拟生成的用户表、物料表、Label表、序列特征表和行为表进行示例说明。您需要在DataWorks中执行SQL命令，将以上表从pai_online_project项目同步到自己的MaxCompute中。具体操作步骤如下：

1. 登录DataWorks控制台。

2. 在左侧导航栏单击数据建模与开发 > 数据开发。

3. 选择已创建的DataWorks工作空间后，单击进入数据开发。

4. 鼠标悬停至新建，选择新建节点 > MaxCompute > ODPS SQL，在弹出的页面中配置节点参数。

   参数	取值建议
   引擎实例	选择已创建的MaxCompute引擎。
   节点类型	ODPS SQL
   路径	业务流程/Workflow/MaxCompute
   名称	可自定义名称。

5. 单击确认。

6. 在新建节点区域运行以下SQL命令，将用户表、物料表、Label表、序列特征表和行为表从pai_online_project项目同步到自己的MaxCompute中。资源组选择已创建的独享资源组。

   同步用户表：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'

   运行后可以获取以下三个分区的数据：

   • ds=20231022

   • ds=20231023

   • ds=20231024

   同步物料表：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'

   运行后可以获取以下三个分区的数据：

   • ds=20231022

   • ds=20231023

   • ds=20231024

   同步Label表：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'

   运行后可以获取以下三个分区的数据：

   • ds=20231022

   • ds=20231023

   • ds=20231024

   同步序列特征表：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'

   运行后可以获取以下三个分区的数据：

   • ds=20231022

   • ds=20231023

   • ds=20231024

   同步行为表：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'

完成上述操作后，您可以在自己的工作空间内查看已同步的表，后续的实践操作将以这五张表为例进行说明。

配置数据源

FeatureStore一般需要配置两个数据源：离线数据源（MaxCompute）和在线数据源（Hologres/GraphCompute/TableStore）。本文以MaxCompute和Hologres为例进行说明。

1. 登录PAI控制台，在左侧导航栏单击数据准备>特征平台（FeatureStore）。

2. 选择工作空间后，单击进入FeatureStore。

3. 配置MaxCompute数据源。

   1. 单击管理数据源，进入管理数据源页面后单击新建数据源，在弹出的页面中配置MaxCompute数据源具体参数。

      参数	取值建议
      类型	MaxCompute
      名称	可自定义名称。
      MaxCompute项目名	选择已创建的MaxCompute项目。

   2. 请复制语句并单击立即前往，同步数据至Hologres，在DataWorks执行该语句后，即可完成授权。

      说明

      授权操作需要您的账号拥有admin权限，具体操作详情请参见通过命令管理用户权限或通过控制台（新版）管理用户权限。

   3. 完成后单击提交。

4. 配置Hologres数据源。

   1. 单击管理数据源。进入管理数据源页面后单击新建数据源，在弹出的页面中配置Hologres数据源具体参数。

      参数	取值建议
      类型	Hologres
      名称	可自定义名称。
      实例ID	选择已创建的Hologres实例名称。
      数据库名称	选择已创建的实例数据库。

   2. 完成后单击提交。

   3. 对Hologres进行授权，具体操作详情请参见配置数据源。

特征项目

您可以通过特征平台创建多个项目空间，每个项目空间是独立的。具体操作，请参见配置FeatureStore项目。运行Notebook需要FeatureStore服务端配合运行，开通特征平台后需要配置数据源，具体操作请参见配置数据源。

其中，offline_datasource_id指的是离线数据源ID，online_datasource_id指的是在线数据源ID。

此处以项目名称是fs_movie为例进行说明。

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

运行以下代码获取当前的project并打印其信息。

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

特征实体（FeatureEntity）

特征实体描述了一组相关的特征集合。多个特征视图可以关联一个特征实体。每个实体都会有一个JoinId，通过JoinId可以关联多个特征视图特征。每一个特征视图都有一个主键（索引键）来获取它的特征数据，但是索引键可以和JoinId定义的名称不一样。

一般在推荐系统中，特征只关联user和item两个特征实体，即特征属于user侧或者item侧。此处以创建 user和item两个特征实体为例进行说明。

• 创建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()
  

• 创建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()

特征视图（FeatureView）

FeatureStore是一个专门用来管理和组织特征数据的平台，外部数据需要通过特征视图进入FeatureStore。特征视图定义了数据从哪里来（DataSource）、需要进行哪些预处理或转换操作（如特征工程/Transformation）、特征的数据结构（包含特征名称和类型在内的特征schema）、数据存储的位置（OnlineStore/OfflineStore），并提供特征元信息管理，如主键、事件时间、分区键、特征实体以及有效期设定ttl（默认-1表示永久有效，正数则表示在线查询时会取ttl内的最新特征数据）。

特征视图分为三种类型：Batch FeatureView（离线特征，或者T-1天特征）、Stream FeatureView（实时特征）、Sequence FeatureView（序列特征）。

Batch FeatureView（离线特征视图）

将离线数据注入到FeatureStore的OfflineStore中，并可以根据需求同步至OnlineStore以支持实时查询。一般的离线特征，或者T-1天的特征。

• 注册user侧离线特征表

  1. 将rec_sln_demo_user_table_preprocess_all_feature_v1注册到特征平台。

     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. 将离线数据源中的rec_sln_demo_user_table_preprocess_all_feature_v1中20231023分区的数据同步到在线数据源。

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

  3. 查看任务运行情况。

     user_task.print_summary()

• 注册item侧离线特征表

  1. 将rec_sln_demo_item_table_preprocess_all_feature_v1表注册到特征平台。

     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. 将离线数据源中的rec_sln_demo_item_table_preprocess_all_feature_v1中20231023分区的数据同步到在线数据源。

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

  3. 查看任务运行情况。

     item_task.print_summary()

Sequence FeatureView（实时序列视图）

支持离线写入序列特征，以及查询和读取实时序列特征。一般在推荐场景中，离线序列特征表（记为F1）前期由模拟生成，后续可以直接由线上日志代替。而在线实时序列的查询中，会查询两张在线行为表的数据，一张为T-1天的行为表（记为B1），另外一张为T天当天的实时行为表（记为B2）。B2是实时计算更新的特征，同时查询B1和B2行为表后拼出该用户的特征序列，然后与其他特征合在一起送到模型打分。

在线T-1天的行为表B1一般是由离线T-1天行为表（记为A1）同步，同步过程中，特征平台会自动完成去重等操作。 在线T天的行为表B2目前需要您通过API接口或Flink等阿里云产品写入。

因此在注册实时特征视图过程中，相当于特征平台同时维护管理了四张表：离线序列表F1、离线T-1天的行为表A1、在线T-1天的行为表B1、在线T天的行为表B2。

注册时，只需要提供离线序列表F1，离线T-1天的行为表A1即可，在线行为表和同步去重过程由特征平台完成。

1. 注册实时特征视图。

   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' # 行为表中的 event time 字段名
     item_id = 'item_id' # 行为表中的 item_id 字段名
     event = 'event' # 行为表中的 event 字段名
     # deduplication_method = 1, 表示 ['user_id', 'item_id', 'event'] 去重
     # deduplication_method = 2, 表示 ['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 指的是离线序列表中序列特征字段名，seq_event 指的是行为字段名，online_seq_name 指的是特征平台在线 Go SDK 查出该 user 对应的行为序列 item_ids 后，会以该名称命名。
     # seq_len 指的是序列长度，大于该长度的序列会被截断。
     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. 将离线数据源中rec_sln_demo_behavior_table_preprocess_v3中20231023分区的数据同步到在线数据源。同步时会自动检查是否有前N天分区的数据，如果没有会自动补充，N可以通过 days_to_load指定，默认是30，一般取30即可。

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

3. 查看任务运行情况。

   seq_task.print_summary()

StreamFeatureView

将数据直接写入OnlineStore，并同时同步到OfflineStore。

Label 表注册

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)

获取在线特征

获取在线特征用于排查离线在线一致性，数据分析等需求，目前优先支持 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.get_online_features(join_ids={'user_id':['169898460', '148811946']}, features=['user_id', 'gender', 'age', 'city'])
print("ret_features = ", ret_features_1)

TrainingSet

训练模型时，首先要构造样本表，样本表由Label数据和特征数据组成。在与FeatureStore交互时，Label数据需要由客户提供，并且需要定义要获取的特征名称，然后根据主键进行point-in-time join（存在event_time的情况下）。

# 指定 Label 表
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, '*') # '*' 代表选择全部特征
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)

模型特征

训练模型并部署成服务后，进行业务预测。其中，训练样本可以从上文的train_set获得。

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)