Platform For AI:FeatureStore Python SDK を使用したレコメンデーションシステムの構築

データテーブルの同期

レコメンデーションシナリオでは、通常、ユーザー特徴テーブル、アイテム特徴テーブル、ラベルテーブル、シーケンス特徴テーブル、行動テーブルといったデータテーブルを準備する必要があります。

このチュートリアルを簡潔にするため、pai_online_project MaxCompute プロジェクトにサンプルのユーザー、アイテム、ラベル、シーケンス特徴、行動テーブルを用意しました。DataWorks で SQL コマンドを実行して、これらのテーブルを pai_online_project プロジェクトからご自身の MaxCompute プロジェクトに同期する必要があります。手順は以下の通りです：

1. DataWorks コンソールにログインします。

2. 左側のナビゲーションウィンドウで、[データ開発と O&M] > [データ開発] をクリックします。

3. ご利用の DataWorks ワークスペースを選択し、[データ開発に進む] をクリックします。

4. [新規] にカーソルを合わせ、[新規ノード] > [MaxCompute] > [ODPS SQL] を選択します。表示されたダイアログボックスで、ノードパラメーターを設定します。

   パラメーター	推奨値
   エンジンインスタンス	ご利用の MaxCompute エンジンを選択します。
   ノードタイプ	ODPS SQL
   パス	Business Flow/Workflow/MaxCompute
   名前	カスタム名を指定できます。

5. [確認] をクリックします。

6. ノードエディターで、以下の SQL コマンドを実行して、ユーザー、アイテム、ラベル、シーケンス特徴、行動テーブルを 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';

   コマンドを実行すると、以下の 3 つのパーティションからデータが返されます：

   • 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';

   コマンドを実行すると、以下の 3 つのパーティションからデータが返されます：

   • ds=20231022

   • ds=20231023

   • ds=20231024

   ラベルテーブルの同期：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'

   コマンドを実行すると、以下の 3 つのパーティションからデータが返されます：

   • 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';

   コマンドを実行すると、以下の 3 つのパーティションからデータが返されます：

   • 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';

これらの手順を完了すると、同期されたテーブルがワークスペースで利用可能になります。これら 5 つのテーブルは、以降のセクションで例として使用されます。

データソースの設定

通常、FeatureStore には 2 つのデータソースを設定する必要があります：オフラインデータソース (MaxCompute) とオンラインデータソース (FeatureDB、Hologres、または TableStore) です。このトピックでは、MaxCompute と FeatureDB を例として使用します。

1. PAI コンソールにログインします。左側のナビゲーションウィンドウで、Data Preparation > FeatureStore をクリックします。

2. ワークスペースを選択し、[フィーチャーストアに進む] をクリックします。

3. MaxCompute データソースを設定します。

   1. [データソース] タブで、Create Store をクリックします。表示されたダイアログボックスで、MaxCompute データソースのパラメーターを設定します。

      パラメーター	推奨値
      Type	MaxCompute
      Name	カスタム名を入力します。
      MaxCompute Project Name	ご利用の MaxCompute プロジェクトを選択します。

   2. 設定が完了したら、Submit をクリックします。

4. FeatureDB データソースを設定します。

   1. すでに FeatureDB データソースを作成している場合は、この手順をスキップできます。

   2. Store タブで、Create Store をクリックします。表示されたダイアログボックスで、FeatureDB データソースのパラメーターを設定します。

      パラメーター	推奨値
      Type	FeatureDB (初めて使用する場合は、画面の指示に従って FeatureDB を有効化してください)
      Name	カスタム名はサポートされていません。デフォルト値は feature_db です。
      Username	ユーザー名を設定します。
      Password	パスワードを設定します。
      VPC 高速接続 (オプション)	設定が成功すると、VPC 内の FeatureStore SDK を使用して、PrivateLink 接続経由で FeatureDB に直接アクセスできます。これにより、データの読み書きパフォーマンスが向上し、アクセスレイテンシが削減されます。
      VPC	オンライン FeatureStore サービスが配置されている VPC を選択します。
      ゾーンと vSwitch	ゾーンと vSwitch を選択します。オンラインサービスのマシンが配置されているゾーンの vSwitch を選択してください。サービスの高い可用性と安定性を確保するために、少なくとも 2 つのゾーンの vSwitch を選択することを推奨します。

   3. 設定が完了したら、Submit をクリックします。

特徴エンジニアリングプロジェクト

FeatureStore では、複数の独立したプロジェクトを作成できます。詳細については、「FeatureStore プロジェクトの設定」をご参照ください。Notebook を実行するには、FeatureStore のサーバーサイド環境が必要です。FeatureStore を有効化した後、データソースも設定する必要があります。詳細については、「新しいデータソースの作成」をご参照ください。

`offline_datasource_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 = fs.get_project(cur_project_name)
print(project)

特徴エンティティ (FeatureEntity)

特徴エンティティは、関連する特徴のセットを記述します。単一の特徴エンティティは、複数の特徴ビューに関連付けることができます。各エンティティには、これらのビューからの特徴を関連付けるために使用される JoinId があります。各特徴ビューには、その特徴データを取得するためのプライマリキー (インデックスキー) がありますが、インデックスキーの名前は JoinId で定義された名前と異なる場合があります。

レコメンデーションシステムでは、特徴は通常、ユーザーとアイテムの 2 つの特徴エンティティにのみ関連付けられます。つまり、特徴はユーザーまたはアイテムのいずれかに属します。このトピックでは、ユーザーとアイテムの特徴エンティティを作成する例を示します。

• ユーザーエンティティの作成

  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_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 に取り込まれます。特徴ビューは、データソース、必要な前処理または変換操作 (特徴エンジニアリングや変換など)、特徴のデータ構造 (特徴名と型を含む)、および保存場所 (オンラインストアまたはオフラインストア) を定義します。また、プライマリキー、イベント時間、パーティションキー、特徴エンティティ、生存時間 (TTL) などの特徴メタデータも管理します。TTL が -1 (デフォルト) の場合は、データが永続的に保存されることを示します。正の数の場合は、オンラインクエリが指定された TTL 内の最新の特徴データを取得することを意味します。

特徴ビューには、バッチ特徴ビュー (オフライン特徴または T-1 日特徴用)、ストリーム特徴ビュー (リアルタイム特徴用)、シーケンス特徴ビュー (シーケンス特徴用) の 3 種類があります。

バッチ特徴ビュー (オフライン特徴ビュー)

バッチ特徴ビューは、オフラインデータを FeatureStore のオフラインストアに注入し、データをオンラインストアに同期してリアルタイムクエリをサポートできます。このタイプのビューは、通常、オフライン特徴または T-1 日特徴に使用されます。

• ユーザー側のオフライン特徴テーブルの登録

  1. rec_sln_demo_user_table_preprocess_all_feature_v1 テーブルを FeatureStore に登録します。

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

• アイテム側のオフライン特徴テーブルの登録

  1. rec_sln_demo_item_table_preprocess_all_feature_v1 テーブルを FeatureStore に登録します。

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

シーケンス特徴ビュー (リアルタイムシーケンスビュー)

シーケンス特徴ビューは、オフラインシーケンス特徴の書き込みとリアルタイムシーケンス特徴のクエリをサポートします。典型的なレコメンデーションシナリオでは、オフラインシーケンス特徴テーブル (F1) は最初にシミュレーションデータから生成され、後でオンラインログに置き換えることができます。オンラインのリアルタイムシーケンスクエリ中、データは 2 つのオンライン行動テーブルからクエリされます：T-1 日の行動テーブル (B1) と T 日のリアルタイム行動テーブル (B2) です。B2 にはリアルタイムで更新される特徴が含まれています。B1 と B2 テーブルからデータがクエリされた後、ユーザーの特徴シーケンスが構築されます。このシーケンスは他の特徴と組み合わされ、スコアリングのためにモデルに送信されます。

オンラインの T-1 日行動テーブル (B1) は、通常、オフラインの T-1 日行動テーブル (A1) から同期されます。同期中、FeatureStore は重複排除などの操作を自動的に実行します。API 操作または Flink などの他の Alibaba Cloud プロダクトを使用して、オンラインの T 日行動テーブル (B2) にデータを書き込む必要があります。

したがって、リアルタイム特徴ビューを登録すると、FeatureStore は同時に 4 つのテーブルを管理します：オフラインシーケンステーブル (F1)、オフライン T-1 日行動テーブル (A1)、オンライン T-1 日行動テーブル (B1)、およびオンライン T 日行動テーブル (B2) です。

登録時には、オフラインシーケンステーブル (F1) とオフライン T-1 日行動テーブル (A1) を提供するだけで済みます。FeatureStore は、オンライン行動テーブルと同期および重複排除プロセスを処理します。

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' # 行動テーブルのイベント時間フィールドの名前。
     item_id = 'item_id' # 行動テーブルの item_id フィールドの名前。
     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 は、FeatureStore オンライン Go SDK によって取得されるユーザーの item_id のシーケンスに使用される名前です。
     # 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 で、ほとんどのシナリオで十分です。

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

3. タスクの実行ステータスを表示します。

   seq_task.print_summary()

ストリーム特徴ビュー (リアルタイム特徴ビュー)

OnlineStore に直接データを書き込み、同時に OfflineStore に同期することは、製品価格や販売数量の更新など、リアルタイムの特徴更新が必要なシナリオに最適です。

ラベルテーブルの登録

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)

トレーニングセット

モデルをトレーニングするときは、まずサンプルテーブルを構築する必要があります。サンプルテーブルは、ラベルデータと特徴データで構成されます。FeatureStore と対話するには、ラベルデータを提供し、取得したい特徴の名前を定義する必要があります。その後、プライマリキーとイベント時間 (存在する場合) に基づいてポイントインタイム結合が実行されます。

# Specify the label table.
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)