Platform For AI:Créer des systèmes de recommandation avec le SDK Python FeatureStore

Synchroniser les tables de données

Un scénario de recommandation nécessite généralement les tables de données suivantes : une table de fonctionnalités utilisateur, une table d'articles, une table d'étiquettes, une table de fonctionnalités de séquence et une table de comportement.

Ce tutoriel pratique utilise des tables de données simulées préparées dans le projet MaxCompute pai_online_project. Pour les utiliser, exécutez des commandes SQL dans DataWorks afin de synchroniser les tables vers votre propre projet MaxCompute. Suivez les étapes ci-dessous :

1. Connectez-vous à la console DataWorks.

2. Dans le volet de navigation de gauche, cliquez sur Data Development & O&M > Data Development.

3. Sélectionnez votre espace de travail DataWorks et cliquez sur Enter Data Development.

4. Placez le curseur sur New, puis choisissez New Node > MaxCompute > ODPS SQL. Configurez les paramètres du nœud comme suit :

   Paramètre	Valeur recommandée
   engine instance	Sélectionnez le moteur MaxCompute que vous avez créé.
   node type	ODPS SQL
   path	Business Flow/Workflow/MaxCompute
   name	Saisissez un nom personnalisé.

5. Cliquez sur Confirm.

6. Dans l'éditeur de nœuds, exécutez les commandes SQL suivantes pour synchroniser les tables de données vers votre projet MaxCompute. Pour Resource Group, sélectionnez votre groupe de ressources exclusif.

   Table utilisateur : 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';

   Une fois la commande terminée, la table contiendra les données des partitions suivantes :

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Table d'articles : 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';

   Une fois la commande terminée, la table contiendra les données des partitions suivantes :

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Table d'étiquettes : 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'

   Une fois la commande terminée, la table contiendra les données des partitions suivantes :

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Table de fonctionnalités de séquence : 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';

   Une fois la commande terminée, la table contiendra les données des partitions suivantes :

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Table de comportement : 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';

Vous pouvez désormais consulter les tables synchronisées dans votre espace de travail. La suite de ce tutoriel utilise ces cinq tables.

Configurer les sources de données

FeatureStore nécessite généralement deux types de sources de données : une source de données hors ligne, telle que MaxCompute, et une source de données en ligne, telle que FeatureDB, Hologres ou Tablestore. Ce tutoriel utilise MaxCompute et FeatureDB à titre d'exemple.

1. Connectez-vous à la console PAI. Dans le volet de navigation de gauche, choisissez Data Preparation > FeatureStore.

2. Sélectionnez un espace de travail et cliquez sur Enter FeatureStore.

3. Configurez une source de données MaxCompute.

   1. Dans l'onglet Create Store, cliquez sur Create Store, puis configurez les paramètres de la source de données MaxCompute dans la page qui s'affiche.

      Paramètre	Valeur recommandée
      Type	MaxCompute
      Name	Saisissez un nom personnalisé.
      MaxCompute Project Name	Sélectionnez le projet MaxCompute que vous avez créé.

   2. Cliquez sur Submit.

4. Configurez une source de données FeatureDB.

   1. Si vous avez déjà créé une source de données FeatureDB, vous pouvez ignorer cette étape.

   2. Dans l'onglet Store, cliquez sur Create Store et configurez les paramètres de la source de données FeatureDB dans la boîte de dialogue.

      Paramètre	Valeur recommandée
      Type	FeatureDB (si vous utilisez FeatureDB pour la première fois, suivez les instructions à l'écran pour l'activer).
      Name	Les noms personnalisés ne sont pas pris en charge. Le nom par défaut est feature_db.
      Username	Définissez un nom d'utilisateur.
      Password	Définissez un mot de passe.
      VPC network high-speed connection (optional)	Cette option vous permet d'utiliser le SDK FeatureStore pour accéder directement à FeatureDB depuis un VPC via un réseau privé. Cette connexion améliore les performances de lecture/écriture des données et réduit la latence d'accès.
      VPC	Sélectionnez le VPC qui contient vos services FeatureStore en ligne.
      Zone and vSwitch	Sélectionnez une zone et un vSwitch. Assurez-vous que le vSwitch se trouve dans la même zone que vos services en ligne. Pour une haute disponibilité, nous vous recommandons de sélectionner des vSwitch dans au moins deux zones.

   3. Cliquez sur Submit.

! pip install https://feature-store-py.oss-cn-beijing.aliyuncs.com/package/feature_store_py-2.2.0-py3-none-any.whl

import unittest
import sys
import os
from os.path import dirname, join, abspath
from feature_store_py.fs_client import FeatureStoreClient
from feature_store_py.fs_project import FeatureStoreProject
from feature_store_py.fs_datasource import UrlDataSource, MaxComputeDataSource, DatahubDataSource, HologresDataSource, SparkDataSource, LabelInput, TrainingSetOutput
from feature_store_py.fs_type import FSTYPE
from feature_store_py.fs_schema import OpenSchema, OpenField
from feature_store_py.fs_feature_view import FeatureView
from feature_store_py.fs_features import FeatureSelector
from feature_store_py.fs_config import LabelInputConfig, PartitionConfig, FeatureViewConfig, TrainSetOutputConfig, SequenceFeatureConfig, SequenceTableConfig
import logging
logger = logging.getLogger("foo")
logger.addHandler(logging.StreamHandler(stream=sys.stdout))

Projet FeatureStore

Vous pouvez créer plusieurs projets dans FeatureStore, chaque projet servant d'espace de noms isolé. Pour obtenir des instructions détaillées, consultez la rubrique Configurer un projet FeatureStore. L'exécution de ce notebook nécessite un service backend FeatureStore et des sources de données configurées. Après avoir activé FeatureStore, vous devez configurer une source de données. Pour plus d'informations, consultez la rubrique Créer une source de données.

Le paramètre offline_datasource_id spécifie l'ID de la source de données hors ligne, et online_datasource_id spécifie l'ID de la source de données en ligne.

Cet exemple utilise un projet nommé 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.")

Exécutez le code suivant pour récupérer le projet actuel et afficher ses informations.

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

Entité de fonctionnalité

Une entité de fonctionnalité décrit un ensemble de fonctionnalités connexes. Plusieurs vues de fonctionnalités peuvent être associées à une seule entité de fonctionnalité. Chaque entité possède un join_id qui permet de joindre les fonctionnalités provenant de différentes vues de fonctionnalités. Chaque vue de fonctionnalités dispose d'une clé primaire pour récupérer ses données de fonctionnalité, et le nom de la clé primaire peut différer du join_id.

Dans les systèmes de recommandation, les fonctionnalités sont généralement associées soit à une entité utilisateur, soit à une entité article. Cet exemple montre comment créer une entité user et une entité item.

• Créer l'entité utilisateur

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

• Créer l'entité article

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

Vue de fonctionnalité

Une vue de fonctionnalité est un objet central dans FeatureStore qui ingère des données externes. Elle définit l'origine des données (la source de données), les opérations de prétraitement ou de transformation éventuelles, la structure des données (le schéma de fonctionnalité, incluant les noms et types de fonctionnalités) ainsi que l'emplacement de stockage (magasin en ligne et/ou magasin hors ligne). Une vue de fonctionnalité gère également les métadonnées, telles que la clé primaire, l'heure de l'événement, la clé de partition, l'entité de fonctionnalité associée et le TTL (Time to Live). Un TTL par défaut de -1 indique que la fonctionnalité est valide en permanence. Une valeur de TTL positive signifie que les requêtes en ligne ne récupèrent que les dernières données de fonctionnalité dans la fenêtre temporelle spécifiée.

Les vues de fonctionnalités sont disponibles en trois types : Batch FeatureView (pour les fonctionnalités hors ligne ou J-1), Stream FeatureView (pour les fonctionnalités en temps réel) et Sequence FeatureView (pour les fonctionnalités de séquence).

Batch FeatureView

Une Batch FeatureView injecte des données hors ligne dans le magasin hors ligne de FeatureStore et peut éventuellement les synchroniser vers le magasin en ligne pour prendre en charge les requêtes en temps réel. Utilisez cette vue de fonctionnalité pour les fonctionnalités hors ligne ou J-1.

• Enregistrer la table de fonctionnalités utilisateur hors ligne

  1. Enregistrez la table rec_sln_demo_user_table_preprocess_all_feature_v1 dans 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. Synchronisez les données de la partition 20231023 de la table rec_sln_demo_user_table_preprocess_all_feature_v1 de la source de données hors ligne vers la source de données en ligne.

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

  3. Vérifiez l'état de la tâche.

     user_task.print_summary()

• Enregistrer la table de fonctionnalités article hors ligne

  1. Enregistrez la table rec_sln_demo_item_table_preprocess_all_feature_v1 dans 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. Synchronisez les données de la partition 20231023 de la table rec_sln_demo_item_table_preprocess_all_feature_v1 de la source de données hors ligne vers la source de données en ligne.

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

  3. Vérifiez l'état de la tâche.

     item_task.print_summary()

Sequence FeatureView

Une Sequence FeatureView prend en charge l'écriture de fonctionnalités de séquence hors ligne et la lecture de fonctionnalités de séquence en temps réel en ligne. Dans les scénarios de recommandation, une table de fonctionnalités de séquence hors ligne (F1) est initialement générée par simulation et peut ensuite être remplacée par des journaux en ligne. Lors des requêtes en ligne pour les séquences en temps réel, les données sont extraites de deux tables de comportement en ligne : une table de comportement J-1 (B1) et une table de comportement en temps réel du jour J (B2). La table B2 contient les fonctionnalités mises à jour par le calcul en temps réel. FeatureStore interroge à la fois B1 et B2, construit la séquence de fonctionnalités de l'utilisateur, puis la combine avec d'autres fonctionnalités pour l'évaluation du modèle.

La table de comportement en ligne J-1 (B1) est généralement synchronisée à partir d'une table de comportement hors ligne J-1 (A1), FeatureStore gérant automatiquement les opérations telles que la déduplication. La table de comportement en ligne du jour J (B2) nécessite actuellement que vous écriviez les données via une API ou d'autres produits Alibaba Cloud tels que Flink.

Par conséquent, lorsque vous enregistrez une Sequence FeatureView, FeatureStore gère quatre tables pour vous : la table de séquence hors ligne (F1), la table de comportement hors ligne J-1 (A1), la table de comportement en ligne J-1 (B1) et la table de comportement en ligne du jour J (B2).

Lors de l'enregistrement, vous fournissez uniquement la table de séquence hors ligne (F1) et la table de comportement hors ligne J-1 (A1). FeatureStore crée les tables en ligne et gère la synchronisation et la déduplication.

1. Enregistrez la Sequence FeatureView.

   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' # Field name for event time in the behavior table.
     item_id = 'item_id' # Field name for item_id in the behavior table.
     event = 'event' # Field name for the event in the behavior table.
     # deduplication_method = 1: Deduplicates by ['user_id', 'item_id', 'event'].
     # deduplication_method = 2: Deduplicates by ['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: Column name of the sequence feature in the offline sequence table.
     # seq_event: The behavior field name used for filtering.
     # online_seq_name: The name under which the corresponding item_id sequence for the user is returned by the online Go SDK.
     # seq_len: The sequence length. Sequences longer than this value are truncated.
     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. Synchronisez les données de la partition 20231023 de la table rec_sln_demo_behavior_table_preprocess_v3 de la source de données hors ligne vers la source de données en ligne. Lors de la synchronisation, le système vérifie automatiquement la présence des données des N jours précédents et les complète si elles sont manquantes. Vous pouvez spécifier N à l'aide du paramètre days_to_load dont la valeur par défaut est 30. Nous vous recommandons d'utiliser la valeur par défaut dans la plupart des cas.

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

3. Vérifiez l'état de la tâche.

   seq_task.print_summary()

Stream FeatureView

Une Stream FeatureView écrit les données directement dans le magasin en ligne et les synchronise simultanément vers le magasin hors ligne. Utilisez-la pour les scénarios nécessitant des mises à jour de fonctionnalités en temps réel, comme la mise à jour des prix ou des volumes de vente des articles.

Enregistrer la table d'étiquettes

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)

Récupérer les fonctionnalités en ligne

Vous pouvez récupérer les fonctionnalités du magasin en ligne à des fins telles que le débogage de la cohérence entre les données hors ligne et en ligne, ainsi que pour l'analyse des données. Cette fonctionnalité est actuellement optimisée pour 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)

Ensemble d'entraînement

Pour entraîner un modèle, commencez par construire un ensemble d'entraînement composé de données d'étiquettes et de fonctionnalités. Lorsque vous interagissez avec FeatureStore, vous fournissez les données d'étiquettes, spécifiez les fonctionnalités à récupérer, puis le système effectue une jointure temporelle (point-in-time join) basée sur la clé primaire (si une heure d'événement existe).

# 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, '*') # '*' selects 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)

Fonctionnalités du modèle

Après avoir entraîné un modèle et l'avoir déployé en tant que service, vous pouvez l'utiliser pour des prédictions métier. Les échantillons d'entraînement sont disponibles dans l'objet train_set de l'étape précédente.

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)

cur_day = '20231024'
pre_day = '20231023'
label_partitions = PartitionConfig(name = 'ds', value = cur_day)
label_input_config = LabelInputConfig(partition_config=label_partitions)

user_partitions = PartitionConfig(name = 'ds', value = pre_day)
feature_view_user_config = FeatureViewConfig(name = 'user_table_preprocess_all_feature_v1',
partition_config=user_partitions)

item_partitions = PartitionConfig(name = 'ds', value = pre_day)
feature_view_item_config = FeatureViewConfig(name = 'item_table_preprocess_all_feature_v1',
partition_config=item_partitions)

seq_partitions = PartitionConfig(name = 'ds', value = cur_day)
feature_view_seq_config = FeatureViewConfig(name = 'wide_seq_feature_v3', partition_config=seq_partitions, event_time='event_unix_time', equal=True)
feature_view_config_list = [feature_view_user_config, feature_view_item_config, feature_view_seq_config]
train_set_partitions = PartitionConfig(name = 'ds', value = cur_day)
train_set_output_config = TrainSetOutputConfig(partition_config=train_set_partitions)

model_name = 'fs_rank_v2'
cur_model = project.get_model(model_name)
task = cur_model.export_train_set(label_input_config, feature_view_config_list, train_set_output_config)
task.wait()
print("task_summary = ", task.task_summary)