人工智能平台 PAI：使用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()

Stream FeatureView（实时特征视图）

将数据直接写入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)