Platform For AI：使用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是Realtime Compute更新的特徵，同時查詢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)