Platform For AI:FeatureStoreを使用してレコメンデーションシステムの機能を管理する

データテーブルの同期

ほとんどのレコメンデーションシナリオでは、user feature table、item feature table、label tableのテーブルを用意する必要があります。

この例では、MaxComputeプロジェクトpai_online_project内の、ユーザーテーブル、アイテムテーブル、およびラベルテーブルを含む3つのシミュレートされたテーブルが使用されます。 ユーザーテーブルとアイテムテーブルの各パーティションには、約100,000のデータレコードが含まれ、MaxComputeプロジェクトでは約70 MBのストレージ容量を占めます。 ラベルテーブルの各パーティションには、約450,000のデータレコードが含まれ、MaxComputeプロジェクトでは約5 MBのストレージ容量を占めます。

DataWorksでSQL文を実行して、ユーザーテーブル、アイテムテーブル、およびラベルテーブルをpai_online_projectプロジェクトから独自のMaxComputeに同期する必要があります。 シミュレートされたテーブルのデータを同期するには、次の手順を実行します。

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

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

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

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.biz日} は、データをバックフィルして次のパーティションのデータを取得するために使用できるパラメーターです。

   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.biz日} は、データをバックフィルして次のパーティションのデータを取得するために使用できるパラメーターです。

   ds=20231022

   ds=20231023

   ds=20231024

   ラベルテーブル: rec_sln_demo_label_tableを同期します。 (クリックして詳細を表示)

   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.biz日} は、データをバックフィルして次のパーティションのデータを取得するために使用できるパラメーターです。

   ds=20231022

   ds=20231023

   ds=20231024

7. データが同期されているテーブルのデータを埋め戻します。

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

   2. 左側のナビゲーションウィンドウで、[自動トリガーノードのO&M] > [自動トリガーノード] を選択します。 [自動トリガーノード] ページが表示されます。

   3. [自動トリガーノード] ページで、管理するノードを見つけ、[操作] 列の [DAG] をクリックします。

   4. 目的のノードを右クリックし、[実行] > [現在のノードのデータの埋め戻し] を選択します。

   5. [データのバックアップ] ダイアログボックスで、[データのタイムスタンプ] を2023-10-22〜2023-10-24に設定し、[OK] をクリックします。

上記の手順を実行した後、ワークスペースでユーザーテーブルrec_sln_demo_user_table_preprocess_all_feature_v1、アイテムテーブルrec_sln_demo_item_table_preprocess_all_feature_v1、およびラベルテーブルrec_sln_demo_label_tableを表示できます。 これらの3つのテーブルは、動作を説明するための例として使用される。

データストアの設定

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

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

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

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

   1. [ストアの管理] をクリックします。 [ストア] ページで、[ストアの作成] をクリックします。 [ストアの作成] パネルで、MaxComputeデータストアの次の表に示すパラメーターを設定します。

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

   2. 承認ステートメントをコピーし、[移動] をクリックしてデータをHologresインスタンスに同期します。 DataWorksでコピーされたステートメントを実行すると、HologresインスタンスはMaxComputeプロジェクトのデータを同期する権限が付与されます。

      説明

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

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

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

   1. [ストアの管理] をクリックします。 [ストア] ページで、[ストアの作成] をクリックします。 ストアの作成パネルで、次の表に示すパラメーターをHologresデータストアに設定します。

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

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

   3. Hologresインスタンスへのアクセス権限を付与します。 詳細については、「データソースの設定」をご参照ください。

Python用FeatureStore SDKのインストール

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

2. 左側のナビゲーションウィンドウで [リソースグループ] をクリックします。

3. [排他的リソースグループ] タブで、管理するリソースグループを見つけます。 [操作] 列のアイコンの上にポインターを移動し、[O&Mアシスタント] を選択します。

4. [コマンドの作成] をクリックします。 [コマンドの作成] パネルで、次の表に示すパラメーターを設定します。

   項目	説明
   Command Name	名前を指定します。 この例では、installが使用されます。
   Command Type	[手動インストール] (pipコマンドを実行してサードパーティパッケージをインストールすることはできません) を選択します。
   Command Content	[command Content] フィールドに次のコマンドを入力します。 /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-1.3.1-py3-none-any.whl
   タイムアウト	タイムアウト期間を指定します。

5. [作成] をクリックします。

6. [コマンドの実行] をクリックします。 表示されるメッセージで、[実行] をクリックします。

7. コマンドの最新のステータスを表示するには、[更新] をクリックします。 コマンドの状態が [成功] に変わると、FeatureStore SDKがインストールされます。

手順1: ルーチンデータ同期ノードの設定

モデルを公開する前に、定期的なデータ同期ノードを設定して、オフラインデータストアからオンラインデータストアにデータを定期的に同期する必要があります。 その後、オンラインデータストアからデータをリアルタイムで読み取ることができます。 この例では、ユーザーテーブルとアイテムテーブルのデータを定期的に同期させる必要があります。 ルーチンデータ同期ノードを設定するには、次の手順を実行します。

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

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

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

4. ユーザーテーブルのデータを定期的に同期します。

   1. ポインタを [作成] に移動し、[ノードの作成] > [MaxCompute] > [PyODPS 3] を選択します。

   2. 次のコードをコードエディターにコピーします。 このコードは、user_table_preprocess_all_feature_v1フィーチャビューからのデータを定期的に同期するために使用されます。

      user_table_preprocess_all_feature_v1フィーチャービューからデータを同期する (クリックして詳細を表示)

      feature_store_py.fs_client importからの

      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フィーチャービューからデータを同期する (クリックして詳細を表示)

      feature_store_py.fs_client importからの

      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 SDK for C ++ と統合されており、低レイテンシで高性能な機能抽出を実装しています。 EasyRecプロセッサがFeatureStore SDK for C ++ を使用してフィーチャを抽出した後、EasyRecプロセッサは抽出されたフィーチャを推論のためにモデルに送信し、スコアをレコメンデーションエンジンに返します。

サービスをデプロイするには、次の手順を実行します。

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

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

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

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",  # The path of the trained model. You can specify a custom path. 
     "model_config": {
       "access_key_id": f'{o.account.access_id}',
       "access_key_secret": f'{o.account.secret_access_key}',
       "region": "cn-beijing",  # Replace the value with the ID of the region in which PAI resides. In this example, cn-beijing is used. 
       "fs_project": "fs_demo",  # Replace the value with the name of your project in FeatureStore. In this example, fs_demo is used. 
       "fs_model": "fs_rank_v1",  # Replace the value with the name of your model feature in FeatureStore. In this example, fs_rank_v1 is used. 
       "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)
   
   # Run the following code if you deploy the service for the first time:
   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")
   
   # Run the following line for routine updates:
   # 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] タブで、デプロイされたサービスを表示します。 詳細については、「PAIコンソールでのモデルサービスのデプロイ」をご参照ください。

10. オプションです。 データストアが存在する仮想プライベートクラウド (VPC) にEASを接続します。 Hologresデータストアなどのデータストアには、指定されたVPC経由でのみアクセスできます。 この例では、Hologresデータストアが使用されています。 Hologresコンソールで、VPC IDやvSwitch IDなどのHologresデータストアに関する基本情報を表示できます。 PAIコンソールのElastic Algorithm Service (EAS) ページの右上隅で、[Configure Direct Connection] をクリックします。 [直接接続の設定] ダイアログボックスで、[VPC] および [vSwitch] フィールドにHologresデータストアのVPC IDとvSwitch IDを入力し、[セキュリティグループ名] パラメーターを設定します。 既存のセキュリティグループを選択するか、新しいセキュリティグループを作成できます。 セキュリティグループで有効になっているポートは、Hologresデータストアへの接続要件を満たしている必要があります。 ほとんどの場合、ポート80はHologresデータストアへの接続に使用されます。 したがって、ポート80が有効になっているセキュリティグループを選択する必要があります。 [OK] をクリックします。 サービスは更新後に利用可能です。

ステップ3: PAI-RECの設定

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

PAI-RECを設定するには、次の手順を実行します。

1. FeatureStoreConfsパラメーターを設定します。

   • RegionId: FeatureStoreが存在するリージョンのID。 この例では、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が使用されます。 パラメーター設定を使用すると、FeatureStore SDK for Goを使用してfs_rank_v1モデルのユーザーフィーチャーエンティティからフィーチャーを抽出PAI-RECます。

       "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パラメーターを設定します。

   AlgoConfsパラメータは、PAI-RECが接続するEASのスコアリングサービスを指定します。

   • 名前: EASを使用してデプロイしたサービスの名前。

   • UrlおよびAuth: EASを使用してデプロイしたサービスにアクセスするために使用されるURLとトークン。 Elastic Algorithm service (EAS) ページでサービス名をクリックし、[サービスの詳細] タブの [エンドポイント情報の表示] をクリックしてURLとトークンを取得します。 詳細については、次をご参照ください： EASについてのFAQ。

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