同步數據表

一般對于推薦場景，需要準備以下數據表：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)