Platform For AI:FeatureStore SDK for Python を使用して推薦システムを構築する

シミュレーションテーブルからのデータ同期

推薦システムには、ユーザー特徴テーブル、アイテム特徴テーブル、ラベルテーブル、シーケンス特徴テーブル、および行動テーブルが必要です。

このトピックでは、MaxCompute プロジェクト pai_online_project のシミュレーションテーブルを例として、FeatureStore SDK for Python の使用方法を説明します。 DataWorks で SQL 文を実行して、pai_online_project プロジェクトからご自身の MaxCompute プロジェクトにシミュレーションテーブルを同期する必要があります。 シミュレーションテーブルを同期するには、次のステップを実行します:

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

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

3. [データ開発] ページで、作成した DataWorks ワークスペースを選択し、[データ開発に進む] をクリックします。

4. [作成] ボタンにポインターを合わせ、[ノードの作成] > [MaxCompute] > [ODPS SQL] を選択します。 [ノードの作成] ダイアログボックスで、次の表で説明されているノードパラメーターを設定します。

   パラメーター	説明
   エンジンインスタンス	作成した MaxCompute エンジンを選択します。
   ノードタイプ	[ノードタイプ] ドロップダウンリストから ODPS SQL を選択します。
   パス	[ビジネスフロー] > [ワークフロー] > [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'

   次のパーティションのデータが同期されます:

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

前述の操作を実行した後、ワークスペースで同期されたテーブルを表示できます。 これらのテーブルは、次のセクションで例として使用されます。

データストアの設定

ほとんどの場合、FeatureStore でオフラインデータストア (MaxCompute プロジェクトなど) とオンラインデータストア (FeatureDB インスタンス、Hologres インスタンス、Tablestore インスタンスなど) を設定する必要があります。 この例では、MaxCompute プロジェクトがオフラインデータストアとして設定され、Hologres インスタンスがオンラインデータストアとして設定されます。

1. PAI コンソールにログインします。 左側のナビゲーションウィンドウで、[データ準備] > [FeatureStore] を選択します。

2. [FeatureStore] ページで、ドロップダウンリストからワークスペースを選択し、[FeatureStore に入る] をクリックします。

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

   1. [ストア] タブで、[ストアの作成] をクリックします。 [ストアの作成] パネルで、次の表で説明されているパラメーターを設定します。

      パラメーター	説明
      タイプ	ドロップダウンリストから [MaxCompute] を選択します。
      名前	データストアの名前を指定します。
      MaxCompute プロジェクト名	作成した MaxCompute プロジェクトを選択します。

   2. 権限付与文をコピーし、[移動] をクリックして DataWorks でコピーした文を実行します。これにより、Hologres インスタンスに、MaxCompute プロジェクトからデータを同期する権限が付与されます。

      説明

      Hologres インスタンスに権限を付与するには、アカウントに管理者権限があることを確認してください。 詳細については、「コマンドを使用したユーザー権限の管理」または「MaxCompute コンソール (新バージョン) でのユーザー権限の管理」をご参照ください。

   3. [送信] をクリックします。

4. Hologres データストアを設定します。

   1. [ストア] タブで、[ストアの作成] をクリックします。 [ストアの作成] パネルで、次の表で説明されているパラメーターを設定します。

      パラメーター	説明
      タイプ	ドロップダウンリストから [Hologres] を選択します。
      名前	データストアの名前を指定します。
      インスタンス ID	作成した Hologres インスタンスを選択します。
      データベース名	Hologres インスタンスで作成したデータベースを選択します。

   2. [送信] をクリックします。

   3. Hologres インスタンスにアクセスするための権限を付与します。 詳細については、「データストアの作成」をご参照ください。

FeatureStore プロジェクト

FeatureStore では複数のプロジェクトを作成できます。 各プロジェクトは独立しています。 詳細については、「FeatureStore プロジェクトの設定」をご参照ください。 Jupyter Notebook でサンプルコードを実行するには、FeatureStore で特定の設定が必要です。 たとえば、FeatureStore でデータストアを設定する必要があります。 詳細については、「データストアの作成」をご参照ください。

このトピックのサンプルコードでは、offline_datasource_id パラメーターはオフラインデータストアの ID を指定し、online_datasource_id パラメーターはオンラインデータストアの ID を指定します。

次のサンプルコードでは、fs_movie プロジェクトを例として、FeatureStore プロジェクトの設定方法を示します。

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)
project.print_summary()

特徴エンティティ

特徴エンティティは、意味的に関連する特徴のコレクションです。 特徴エンティティは、複数の特徴ビューに関連付けることができます。 各エンティティには join ID があります。 join ID を使用して、複数の特徴ビューにまたがる特徴を関連付けることができます。 各特徴ビューには、特徴を取得するために使用できるプライマリキー (インデックス) があります。 プライマリキーは、join ID の名前とは異なる場合があります。

通常、推薦システムには、ユーザーとアイテムの 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()

特徴ビュー

特徴ビューを作成して、外部ソースから FeatureStore に特徴データを取り込み、一元管理できます。 特徴ビューには、データソース、必要な変換、特徴テーブルスキーマ、オンラインおよびオフラインデータストアなど、特徴を管理するために必要なすべての情報が含まれています。 特徴ビューでは、プライマリキー、イベント時間、パーティションフィールド、特徴エンティティ、存続時間 (TTL) などのメタデータも管理できます。 TTL は、オンラインデータストアで特徴データが利用可能な期間を指定するパラメーターです。 デフォルト値の -1 は、オンラインデータストアがすべての特徴データを保持することを指定します。 正の値は、オンラインデータストアが指定された期間内の最新の特徴データのみを保持することを指定します。

FeatureStore は、バッチ特徴ビュー、ストリーム特徴ビュー、シーケンス特徴ビューの各タイプをサポートしています。

バッチ特徴ビュー

バッチ特徴ビューを使用すると、オフライン特徴をオフラインデータストアに取り込み、オフライン特徴をオンラインデータストアに同期できます。 これにより、リアルタイムで特徴をクエリできます。

• ユーザーテーブルを登録します。

  1. FeatureStore にユーザーテーブル 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. オフラインデータストアからオンラインデータストアに ds=20231023 パーティションのデータを同期します。

     user_task = user_feature_view.publish_table({'ds':'20231023'})
     user_task.wait()

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

     user_task.print_summary()

• アイテムテーブルを登録します。

  1. FeatureStore にアイテムテーブル 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. オフラインデータストアからオンラインデータストアに ds=20231023 パーティションのデータを同期します。

     item_task = item_feature_view.publish_table({'ds':'20231023'})
     item_task.wait()

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

     item_task.print_summary()

シーケンス特徴ビュー

シーケンス特徴ビューを使用すると、シーケンス特徴をオフラインで書き込み、リアルタイムのシーケンス特徴をオンラインで読み取ることができます。 推薦システムを構築する際に、シミュレートされたオフラインシーケンステーブル (F1 テーブル) を FeatureStore に登録できます。 シミュレートされたテーブルのデータは、その後オンラインログに置き換えられます。 リアルタイムのシーケンス特徴をクエリするために、システムは当日のオンライン行動テーブル (B2 テーブル) と前日のオンライン行動テーブル (B1 テーブル) からデータを取得します。 これら 2 つのテーブルから取得されたデータは連結されてシーケンス特徴を生成し、スコアリングのためにモデルに送信されます。

B1 テーブルは、前日のオフライン行動テーブル (A1 テーブル) からデータを同期します。 重複データは、同期中に FeatureStore によって自動的にフィルター処理されます。 B2 テーブルには、リアルタイムの行動特徴が含まれています。 API 操作を呼び出すか、Realtime Compute for Apache Flink を使用して、リアルタイムデータを B2 テーブルに書き込むことができます。

シーケンス特徴ビューを作成すると、FeatureStore は F1、A1、B1、および B2 テーブルを一元的に管理します。

オフラインデータストアの F1 テーブルと A1 テーブルのみをシーケンス特徴ビューに登録する必要があります。 FeatureStore は、同期と重複排除によってオンライン行動テーブル (B1 テーブルと B2 テーブル) を自動的に作成します。

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 SDK for Go を使用して取得されるオンラインシーケンス特徴のフィールド名を指定します。
     # 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 の ds=20231023 パーティションのデータを同期します。 同期中に、FeatureStore はデータが欠落している場合に自動的にチェックし、既存データをロードします。days_to_load パラメーターは、FeatureStore がデータの可用性をチェックする過去の日数を指定します。 このパラメーターのデフォルト値は 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)

オンライン特徴の取得

オンライン特徴を取得してデータを分析し、オンラインデータストアとオフラインデータストアで特徴が一致しているかどうかを確認できます。

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, '*') # '*' Specifies all features.
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)

モデル特徴

FeatureStore によって生成されたトレーニングデータセットでモデルをトレーニングし、トレーニング済みのモデルを PAI の推論サービスとしてデプロイできます。

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)