Platform For AI:FeatureStore を使用した推薦システムにおける特徴の管理

データテーブルの同期

一般的な推薦シナリオでは、ユーザー特徴テーブル、アイテム特徴テーブル、ラベルテーブルの 3 つのデータテーブルを準備する必要があります。

このトピックを簡単に理解できるように、MaxCompute の pai_online_project プロジェクトに、シミュレートされたユーザー、アイテム、ラベルのテーブルを準備しました。ユーザーテーブルとアイテムテーブルは、パーティションごとに約 100,000 件のデータエントリがあり、MaxCompute で約 70 MB を占有します。ラベルテーブルは、パーティションごとに約 450,000 件のデータエントリがあり、MaxCompute で約 5 MB を占有します。

DataWorks で SQL コマンドを実行して、pai_online_project プロジェクトからご利用の MaxCompute プロジェクトにユーザー、アイテム、ラベルのテーブルを同期します。手順は次のとおりです：

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

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

3. 作成した DataWorks ワークスペースを選択し、[Data Studio に移動] をクリックします。

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='${bdp.system.bizdate}')
   SELECT * except(ds)
   FROM pai_online_project.rec_sln_demo_user_table_preprocess_all_feature_v1
   WHERE ds = '${bdp.system.bizdate}';

   コード内の ${bdp.system.bizdate} はパラメーターです。これを 20231022、20231023、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='${bdp.system.bizdate}')
   SELECT * except(ds)
   FROM pai_online_project.rec_sln_demo_item_table_preprocess_all_feature_v1
   WHERE ds = '${bdp.system.bizdate}';

   コード内の ${bdp.system.bizdate} はパラメーターです。これを 20231022、20231023、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='${bdp.system.bizdate}')
   SELECT * except(ds)
   FROM pai_online_project.rec_sln_demo_label_table
   WHERE ds = '${bdp.system.bizdate}';

   コード内の ${bdp.system.bizdate} はパラメーターです。これを 20231022、20231023、20231024 に置き換えてから、データバックフィルを実行し、3つのパーティションのデータを取得します。データバックフィルの手順は後述します。

   ds=20231022

   ds=20231023

   ds=20231024

7. 同期したテーブルのデータバックフィルを実行します。

   1. DataWorks コンソールで、左側のナビゲーションウィンドウで [データ開発と O&M] > [オペレーションセンター] をクリックします。ドロップダウンリストから対応するワークスペースを選択し、[オペレーションセンターに入る] をクリックします。

   2. 左側のナビゲーションウィンドウで、[自動トリガータスク O&M] > [自動トリガータスク] をクリックして、[自動トリガータスク] ページに移動します。

   3. 定期タスクのリストで、対象のタスクをクリックして、その有向非循環グラフ (DAG) を開きます。

   4. 対象のノードを右クリックし、[データバックフィル] > [現在のノード] を選択します。データバックフィルモードを選択します。

   5. [データタイムスタンプ] を 2023-10-22 から 2023-10-24 の範囲に設定し、[送信] をクリックします。

これらの手順を完了すると、ワークスペースでユーザーテーブル rec_sln_demo_user_table_preprocess_all_feature_v1、アイテムテーブル rec_sln_demo_item_table_preprocess_all_feature_v1、およびラベルテーブル rec_sln_demo_label_table を表示できます。以降の操作では、これら 3 つのテーブルを例として使用します。

データソースの構成

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

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

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

3. MaxCompute データソースを構成します。

   1. Data Source タブで、Create Store をクリックします。表示されるページで、MaxCompute データソースパラメーターを設定します。

      パラメーター	推奨値
      Type	MaxCompute
      Name	カスタム名を指定できます。
      MaxCompute Project Name	作成した MaxCompute プロジェクトを選択します。

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 Python SDK のインストール

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

2. 左側のナビゲーションウィンドウで、Resource Group をクリックします。

3. [専用リソースグループ] タブで、[目的] が [データスケジューリング] に設定されているリソースグループを見つけます。スケジュールリソースに対応する アイコンをクリックし、[O&M アシスタント] を選択します。

4. [コマンドの作成] をクリックします。表示されたページで、コマンドパラメーターを構成します。

   パラメーター	推奨値
   コマンド名	カスタム名を入力します。このトピックでは install を例として使用します。
   コマンドタイプ	手動入力 (pip コマンドはサードパーティパッケージのインストールには使用できません)
   コマンド内容	/home/tops/bin/pip3 install -i https://pypi.tuna.tsinghua.edu.cn/simple https://feature-store-py.oss-cn-beijing.aliyuncs.com/package/feature_store_py-2.0.2-py3-none-any.whl
   タイムアウト	カスタム時間を設定できます。

5. OK をクリックしてコマンドを作成します。

6. Command をクリックします。表示されたページで、Running をクリックします。

7. ボタンをクリックして最新の実行ステータスを表示します。ステータスが Successful に変わると、インストールは完了です。

ステップ 1：定例のデータノード同期

モデルを公開する前に、データ同期ノードをスケジュールする必要があります。これにより、データがオフラインストアからオンラインストアに定期的に同期され、リアルタイムでの読み取りが可能になります。この例では、ユーザー特徴テーブルとアイテム特徴テーブルの同期をスケジュールします。手順は次のとおりです。

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

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

3. 作成した DataWorks ワークスペースを選択し、[Data Studio に移動] をクリックします。

4. ユーザーテーブルの同期をスケジュールします。

   1. [作成] にカーソルを合わせ、[ノードの作成] > [MaxCompute] > [PyODPS 3] を選択します。

   2. スクリプトに次の内容をコピーして、user_table_preprocess_all_feature_v1 の同期をスケジュールします。

      user_table_preprocess_all_feature_v1 の同期をスケジュール (クリックして詳細を表示)

      from feature_store_py.fs_client import FeatureStoreClient
      import datetime
      from feature_store_py.fs_datasource import MaxComputeDataSource
      import sys
      
      cur_day = args['dt']
      print('cur_day = ', cur_day)
      
      access_key_id = o.account.access_id
      access_key_secret = o.account.secret_access_key
      fs = FeatureStoreClient(access_key_id=access_key_id, access_key_secret=access_key_secret, region='cn-beijing')
      cur_project_name = 'fs_demo'
      project = fs.get_project(cur_project_name)
      
      feature_view_name = 'user_table_preprocess_all_feature_v1'
      batch_feature_view = project.get_feature_view(feature_view_name)
      task = batch_feature_view.publish_table(partitions={'ds':cur_day}, mode='Overwrite')
      task.wait()
      task.print_summary()

   3. 右側のペインで、[スケジューリング構成] をクリックします。表示されたページで、スケジューリングパラメーターを構成します。

      パラメーター		推奨値
      スケジューリングパラメーター	パラメーター名	dt
      	パラメーター値	$[yyyymmdd-1]
      リソースプロパティ	スケジューリングリソースグループ	作成した専用スケジューリングリソースグループを選択します。
      スケジューリング依存関係		作成したユーザーテーブルを選択します。

   4. ノードを構成してテストした後、ノード構成を保存して送信します。

   5. データバックフィルを実行します。詳細については、「データテーブルの同期」をご参照ください。

5. アイテムテーブルの同期をスケジュールします。

   1. [作成] にカーソルを合わせ、[ノードの作成] > [MaxCompute] > [PyODPS 3] を選択します。表示されたページで、ノードパラメーターを構成します。

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

   3. スクリプトに次の内容をコピーします。

      item_table_preprocess_all_feature_v1 の同期をスケジュール (クリックして詳細を表示)

      from feature_store_py.fs_client import FeatureStoreClient
      import datetime
      from feature_store_py.fs_datasource import MaxComputeDataSource
      import sys
      
      cur_day = args['dt']
      print('cur_day = ', cur_day)
      
      access_key_id = o.account.access_id
      access_key_secret = o.account.secret_access_key
      fs = FeatureStoreClient(access_key_id=access_key_id, access_key_secret=access_key_secret, region='cn-beijing')
      cur_project_name = 'fs_demo'
      project = fs.get_project(cur_project_name)
      
      feature_view_name = 'item_table_preprocess_all_feature_v1'
      batch_feature_view = project.get_feature_view(feature_view_name)
      task = batch_feature_view.publish_table(partitions={'ds':cur_day}, mode='Overwrite')
      task.wait()
      task.print_summary()

   4. 右側のペインで、[スケジューリング構成] をクリックします。表示されたページで、スケジューリングパラメーターを構成します。

      パラメーター		推奨値
      スケジューリングパラメーター	パラメーター名	dt
      	パラメーター値	$[yyyymmdd-1]
      リソースプロパティ	スケジューリングリソースグループ	作成した専用スケジューリングリソースグループを選択します。
      スケジューリング依存関係		作成したアイテムテーブルを選択します。

   5. ノードを構成してテストした後、ノード構成を保存して送信します。

   6. データバックフィルを実行します。詳細については、「データテーブルの同期」をご参照ください。

6. 同期が完了した後、Hologres で最新の同期された特徴を表示できます。

ステップ 2：EAS モデルサービスの作成とデプロイ

モデルサービスは、推薦エンジンからのリクエストを受け取り、リクエストに基づいて対応するアイテムをスコアリングし、そのスコアを返します。EasyRec プロセッサには、低遅延でパフォーマンス専有型の特徴フェッチを実現する FeatureStore Cpp SDK が含まれています。プロセッサは特徴をフェッチし、それをモデルに送信して推論を行い、結果のスコアを推薦エンジンに返します。

モデルサービスをデプロイする手順は次のとおりです。

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

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

3. 作成した DataWorks ワークスペースを選択し、[Data Studio に移動] をクリックします。

4. [作成] にカーソルを合わせ、[ノードの作成] > [MaxCompute] > [PyODPS 3] を選択します。

5. スクリプトに次の内容をコピーします。

   import os
   import json
   config = {
     "name": "fs_demo_v1",
     "metadata": {
       "cpu": 4,
       "rpc.max_queue_size": 256,
       "rpc.enable_jemalloc": 1,
       "gateway": "default",
       "memory": 16000
     },
     "model_path": f"oss://beijing0009/EasyRec/deploy/rec_sln_demo_dbmtl_v1/{args['ymd']}/export/final_with_fg",  # トレーニング済みモデルのパス。パスはカスタマイズできます。
     "model_config": {
       "access_key_id": f'{o.account.access_id}',
       "access_key_secret": f'{o.account.secret_access_key}',
       "region": "cn-beijing",  # PAI がデプロイされているリージョンに置き換えてください。このトピックでは cn-beijing を例として使用します。
       "fs_project": "fs_demo",  # ご利用の FeatureStore プロジェクトの名前に置き換えてください。このトピックでは fs_demo を例として使用します。
       "fs_model": "fs_rank_v1",  # ご利用の FeatureStore モデル特徴の名前に置き換えてください。このトピックでは fs_rank_v1 を例として使用します。
       "fs_entity": "item",
       "load_feature_from_offlinestore": True,
       "steady_mode": True,
       "period": 2880,
       "outputs": "probs_is_click,y_ln_playtime,probs_is_praise",
       "fg_mode": "tf"
     },
     "processor": "easyrec-1.9",
     "processor_type": "cpp"
   }
   
   with open("echo.json", "w") as output_file:
       json.dump(config, output_file)
   
   # 初回デプロイメントの場合は次の行を実行します
   os.system(f"/home/admin/usertools/tools/eascmd -i {o.account.access_id} -k {o.account.secret_access_key} -e pai-eas.cn-beijing.aliyuncs.com create echo.json")
   
   # 定期的な更新の場合は次の行を実行します
   # os.system(f"/home/admin/usertools/tools/eascmd -i {o.account.access_id} -k {o.account.secret_access_key} -e pai-eas.cn-beijing.aliyuncs.com modify fs_demo_v1 -s echo.json")

6. 右側のペインで、[スケジューリング構成] をクリックします。表示されたページで、スケジューリングパラメーターを構成します。

   パラメーター		推奨値
   スケジューリングパラメーター	パラメーター名	dt
   	パラメーター値	$[yyyymmdd-1]
   リソースプロパティ	スケジューリングリソースグループ	作成した専用スケジューリングリソースグループを選択します。
   スケジューリング依存関係		対応するトレーニングタスクと item_table_preprocess_all_feature_v1 を選択します。

7. ノードを構成してテストした後、実行してデプロイメントステータスを確認します。

8. デプロイメントが完了したら、34 行目をコメントアウトし、37 行目のコメントを解除して、タスクを定期実行のために送信します。

9. (オプション) Elastic Algorithm Service (EAS) ページの Inference Service タブで、デプロイされたサービスを表示できます。詳細については、「カスタムデプロイメント」をご参照ください。

10. (オプション) Hologres のように特定の VPC を通じてのみアクセス可能なデータソースを使用する場合、EAS ネットワークをデータソースの VPC に接続する必要があります。たとえば、Hologres を使用する場合、Hologres インスタンスの [ネットワーク情報] ページで VPC ID と vSwitch ID を見つけることができます。EAS サービスページで、右上隅の [高速接続の構成] をクリックし、VPC ID と vSwitch ID を入力します。また、既存のセキュリティグループを選択するか、新しいセキュリティグループを作成して セキュリティグループ名 を提供する必要もあります。セキュリティグループが Hologres で必要なポートでのトラフィックを許可していることを確認してください。Hologres 接続は通常ポート 80 を使用するため、選択したセキュリティグループはポート 80 でのトラフィックを許可する必要があります。すべての情報を入力した後、[OK] をクリックします。サービスが更新された後、使用できます。

ステップ 3：PAI-REC の構成

PAI-REC は、FeatureStore Go SDK を統合し、FeatureStore および EAS とシームレスに接続できる推薦エンジンサービスです。

構成手順は次のとおりです。

1. FeatureStoreConfs の構成

   • RegionId：プロダクトが配置されているリージョン。このトピックでは cn-beijing を例として使用します。

   • ProjectName：作成した FeatureStore プロジェクトの名前、つまり fs_demo です。

       "FeatureStoreConfs": {
           "pairec-fs": {
               "RegionId": "cn-beijing",
               "AccessId": "${AccessKey}",
               "AccessKey": "${AccessSecret}",
               "ProjectName": "fs_demo"
           }
       },

2. FeatureConfs の構成

   • FeatureStoreName：前のステップの FeatureStoreConfs の pairec-fs 設定と同じでなければなりません。

   • FeatureStoreModelName：作成したモデル特徴の名前、つまり fs_rank_v1 です。

   • FeatureStoreEntityName：作成した特徴エンティティの名前、つまり user です。これにより、PAI-REC の FeatureStore Go SDK は fs_rank_v1 モデルの user エンティティから特徴をフェッチするように指示されます。

       "FeatureConfs": {
           "recreation_rec": {
               "AsynLoadFeature": true,
               "FeatureLoadConfs": [
                   {
                       "FeatureDaoConf": {
                           "AdapterType": "featurestore",
                           "FeatureStoreName": "pairec-fs",
                           "FeatureKey": "user:uid",
                           "FeatureStoreModelName": "fs_rank_v1",
                           "FeatureStoreEntityName": "user",
                           "FeatureStore": "user"
                       }
                   }
               ]
           }
       },

3. AlgoConfs の構成

   この構成は、PAI-REC がどの EAS モデルスコアリングサービスに接続するかを指示します。

   • Name：デプロイされた EAS サービスの名前と同じでなければなりません。

   • Url および Auth: EAS サービスによって提供される URL とトークンです。この情報を取得するには、EAS モデルサービスページでサービス名をクリックします。次に、Overview タブの Basic Information セクションで、View Endpoint Information をクリックします。詳細な構成については、「EAS よくある質問」をご参照ください。

       "AlgoConfs": [
           {
               "Name": "fs_demo_v1",
               "Type": "EAS",
               "EasConf": {
                   "Processor": "EasyRec",
                   "Timeout": 300,
                   "ResponseFuncName": "easyrecMutValResponseFunc",
                   "Url": "eas_url_xxx",
                   "EndpointType": "DIRECT",
                   "Auth": "eas_token"
               }
           }
       ],