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

データテーブルの同期

レコメンデーションシナリオでは、通常、ユーザー特徴テーブル、アイテムテーブル、ラベルテーブル、シーケンス特徴テーブル、行動テーブルなどのデータテーブルが必要です。

このハンズオンチュートリアルでは、pai_online_project MaxCompute プロジェクトで用意されているシミュレーションデータテーブルを使用します。これらのテーブルを使用するには、DataWorks で SQL コマンドを実行し、テーブルをご自身の MaxCompute プロジェクトに同期します。手順は次のとおりです。

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

2. 左側メニューで、データ開発 & O&M > データ開発 をクリックします。

3. DataWorks ワークスペースを選択し、データ開発に入る をクリックします。

4. 新規にカーソルを合わせ、新しいノード > MaxCompute > ODPS SQL を選択します。ノードのパラメータを次のように設定します。

   パラメータ	推奨値
   エンジンインスタンス	作成した MaxCompute エンジンを選択します。
   ノードタイプ	ODPS SQL
   パス	Business Flow/Workflow/MaxCompute
   名前	カスタム名を入力します。

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

6. ノードエディタで、次の SQL コマンドを実行してデータテーブルを 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

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

これで、ワークスペースで同期したテーブルを表示できます。本チュートリアルのこれ以降では、これら5つのテーブルを使用します。

データソースの設定

FeatureStore では通常、オフラインデータソース (MaxCompute など) とオンラインデータソース (FeatureDB、Hologres、Tablestore など) の 2 種類のデータソースが必要です。このチュートリアルでは、MaxCompute と FeatureDB を例として使用します。

1. PAI コンソールにログインします。左側メニューで、[データ準備] > [フィーチャーストア (Feature Store)] を選択します。

2. ワークスペースを選択し、FeatureStore に入る をクリックします。

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

   1. [データソースの新規作成] タブで [データソースの新規作成] をクリックし、表示されたページで MaxCompute データソースのパラメータを設定します。

      パラメータ	推奨値
      [タイプ]	MaxCompute
      [名前]	カスタム名を入力します。
      [MaxCompute プロジェクト名]	作成した MaxCompute プロジェクトを選択します。

   2. [コミット] をクリックします。

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

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

   2. [データソース] タブで [データソースの新規作成] をクリックし、ダイアログボックスで FeatureDB データソースのパラメータを設定します。

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

   3. [コミット] をクリックします。

FeatureStore プロジェクト

FeatureStore では複数のプロジェクトを作成でき、各プロジェクトは独立した名前空間として機能します。詳細な手順については、「FeatureStore プロジェクトを設定する」をご参照ください。このノートブックを実行するには、バックエンドの FeatureStore サービスと設定済みのデータソースが必要です。FeatureStore をアクティブ化した後、データソースを設定する必要があります。詳細については、「データソースを作成する」をご参照ください。

offline_datasource_id パラメーターはオフラインデータソースの ID を、online_datasource_id はオンラインデータソースの ID を指定します。

この例では、fs_demo という名前のプロジェクトを使用します。

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(f"The project '{cur_project_name}' does not exist. You must create it first.")

次のコードを実行して現在のプロジェクトを取得し、その情報を出力します。

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

特徴量エンティティ

フィーチャーエンティティは、関連するフィーチャーの集合です。複数のフィーチャービューを単一のフィーチャーエンティティに関連付けることができます。各エンティティには、異なるフィーチャービューのフィーチャーを結合するための join_id があります。各フィーチャービューには、そのフィーチャーデータを取得するための主キーがあり、主キーの名前は join_id とは異なる場合があります。

レコメンデーションシステムでは、特徴量は通常、ユーザーまたはアイテムのエンティティに関連付けられます。この例では、user エンティティと item エンティティの両方を作成する方法を示します。

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

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

特徴量ビュー

特徴量ビューは、外部データを取り込む FeatureStore のコアオブジェクトです。データの取得元 (データソース)、前処理や変換操作、データ構造 (特徴量名と型を含む特徴量スキーマ)、および保存場所 (オンラインストアおよび/またはオフラインストア) を定義します。特徴量ビューは、プライマリキー、イベントタイム、パーティションキー、関連する特徴量エンティティ、TTL (Time to Live) などのメタデータも管理します。デフォルトの 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 テーブルには、リアルタイム計算によって更新された特徴量が含まれています。FeatureStore は B1 と B2 の両方をクエリし、ユーザーの特徴量シーケンスを構築し、それを他の特徴量と組み合わせてモデルスコアリングを行います。

オンラインの T-1 日の行動テーブル (B1) は通常、オフラインの T-1 日の行動テーブル (A1) から同期され、FeatureStore は重複排除などの操作を自動的に処理します。当日 (T 日) のオンライン行動テーブル (B2) は、現在、API または Flink などの他の Alibaba Cloud 製品を介してデータを書き込む必要があります。

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

登録時には、オフラインシーケンステーブル (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: オンラインの 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()

ストリーム特徴量ビュー

ストリーム特徴量ビューは、データをオンラインストアに直接書き込み、同時にオフラインストアに同期します。アイテムの価格や販売数量の更新など、リアルタイムの特徴量更新が必要なシナリオで使用します。

ラベルテーブルの登録

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.list_feature_view_online_features(join_ids=['169898460', '148811946'])
print("ret_features = ", ret_features_1)

トレーニングセット

モデルをトレーニングするには、まずラベルと特徴量データのトレーニングセットを構築します。FeatureStore と対話する際、ラベルデータを提供し、取得する特徴量を指定すると、システムはプライマリキーに基づいてポイントインタイム結合を実行します (イベントタイムが存在する場合)。

# ラベルテーブルを指定
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)