FeatureStore SDK for Python - Platform For AI - Centre de documentation Alibaba Cloud

Utilisez le kit SDK Python de FeatureStore pour enregistrer des vues de fonctionnalités, écrire des données de fonctionnalités dans les magasins hors ligne et en ligne, et générer des ensembles d'entraînement pour l'entraînement des modèles.

Ce guide utilise le jeu de données open source Moviedata comme exemple pratique. Les tables Movie, User et Rating correspondent respectivement aux tables d'éléments, d'utilisateurs et d'étiquettes dans un pipeline de recommandation classique.

Dans ce guide, vous allez :

Installer le SDK et configurer votre projet
Définir les entités de fonctionnalités et les vues de fonctionnalités
Écrire des données dans le magasin hors ligne et les publier dans le magasin en ligne
Récupérer les fonctionnalités en ligne et définir un sélecteur de fonctionnalités
Créer un ensemble d'entraînement et l'exporter pour l'entraînement du modèle

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Pour une expérience optimale, exécutez le code de ce guide sur une instance DSW.

Étape 1 : Installer le SDK et configurer votre projet

Installer le SDK

Exécutez la commande suivante dans un environnement Python 3 :

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

Définir les variables d'environnement

Stockez vos identifiants sous forme de variables d'environnement afin d'éviter de coder en dur des informations sensibles. Dans votre notebook DSW, cliquez sur Terminal dans la barre de menu supérieure, puis exécutez :

echo "export AccessKeyID='<your-access-key-id>'" >> ~/.bashrc
echo "export AccessKeySecret='<your-access-key-secret>'" >> ~/.bashrc
source ~/.bashrc

Remplacez <your-access-key-id> et <your-access-key-secret> par votre AccessKey ID et votre AccessKey Secret réels.

Importer les modules

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 EASDeployConfig, LabelInputConfig, PartitionConfig, FeatureViewConfig, TrainSetOutputConfig, SequenceFeatureConfig, SequenceTableConfig
import logging
logger = logging.getLogger("foo")
logger.addHandler(logging.StreamHandler(stream=sys.stdout))

Se connecter à un projet FeatureStore

Initialisez le client et connectez-vous à votre projet. Vous pouvez créer plusieurs projets indépendants dans FeatureStore. Ce guide utilise un projet nommé fs_movie.

# Load credentials from environment variables
access_id = os.getenv("AccessKeyID")
access_ak = os.getenv("AccessKeySecret")

# Set the region where your FeatureStore is activated
region = 'cn-hangzhou'

fs = FeatureStoreClient(access_key_id=access_id, access_key_secret=access_ak, region=region)

cur_project_name = "fs_movie"
project = fs.get_project(cur_project_name)
if project is None:
    raise ValueError("Project not found. Create the project first: fs_movie")

Pour vérifier la connexion, affichez les détails du projet :

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

Étape 2 : Définir les entités de fonctionnalités

Une entité de fonctionnalité regroupe des fonctionnalités sémantiquement liées. Chaque entité possède un ID de jointure qui relie les fonctionnalités entre plusieurs vues de fonctionnalités. Les vues de fonctionnalités peuvent utiliser des noms de clé primaire différents, mais elles effectuent la jointure sur cet ID d'entité partagé.

L'exemple Moviedata utilise trois entités :

# Movie entity
cur_entity_name_movie = "movie_data"
entity_movie = project.get_entity(cur_entity_name_movie)
if entity_movie is None:
    entity_movie = project.create_entity(name=cur_entity_name_movie, join_id='movie_id')
entity_movie.print_summary()

# User entity
cur_entity_name_user = "user_data"
entity_user = project.get_entity(cur_entity_name_user)
if entity_user is None:
    entity_user = project.create_entity(name=cur_entity_name_user, join_id='user_md5')
entity_user.print_summary()

# Rating entity
cur_entity_name_ratings = "rating_data"
entity_ratings = project.get_entity(cur_entity_name_ratings)
if entity_ratings is None:
    entity_ratings = project.create_entity(name=cur_entity_name_ratings, join_id='rating_id')
entity_ratings.print_summary()

Étape 3 : Créer des vues de fonctionnalités et écrire des données

Les vues de fonctionnalités définissent la manière dont les données externes entrent dans FeatureStore : la source de données, le schéma, l'emplacement de stockage et les métadonnées des fonctionnalités. FeatureStore prend en charge trois types de vues de fonctionnalités :

Type	Cas d'utilisation	Chemin d'écriture
`BatchFeatureView`	Fonctionnalités hors ligne ou J-1	Écriture dans le magasin hors ligne → publication dans le magasin en ligne
`StreamFeatureView`	Fonctionnalités en temps réel	Écriture directe dans le magasin en ligne
`Sequence FeatureView`	Séquences de comportement utilisateur	Écriture hors ligne → lecture en ligne en temps réel

Le Time to Live (TTL) contrôle la durée de conservation des données dans le magasin en ligne. -1 conserve toutes les données ; une valeur positive ne conserve que les données comprises dans la période spécifiée.

BatchFeatureView

BatchFeatureView gère les fonctionnalités hors ligne ou J-1. Le flux de données suit deux opérations distinctes :

write_table() — écrit les données dans le magasin hors ligne MaxCompute. Cette opération enregistre les données uniquement dans le magasin hors ligne ; elles ne sont pas encore disponibles pour les requêtes en ligne.
publish_table() — synchronise les données du magasin hors ligne vers le magasin en ligne, ce qui les rend disponibles pour la récupération en temps réel.

Vue de fonctionnalités Movie

# Load the movie CSV from a public URL
path = 'https://feature-store-test.oss-cn-beijing.aliyuncs.com/dataset/moviedata_all/movies.csv'
ds = UrlDataSource(path, delimiter=',', omit_header=True)

# Define the schema
movie_schema = OpenSchema(
    OpenField(name='movie_id', type='STRING'),
    OpenField(name='name', type='STRING'),
    OpenField(name='alias', type='STRING'),
    OpenField(name='actors', type='STRING'),
    OpenField(name='cover', type='STRING'),
    OpenField(name='directors', type='STRING'),
    OpenField(name='double_score', type='STRING'),
    OpenField(name='double_votes', type='STRING'),
    OpenField(name='genres', type='STRING'),
    OpenField(name='imdb_id', type='STRING'),
    OpenField(name='languages', type='STRING'),
    OpenField(name='mins', type='STRING'),
    OpenField(name='official_site', type='STRING'),
    OpenField(name='regions', type='STRING'),
    OpenField(name='release_date', type='STRING'),
    OpenField(name='slug', type='STRING'),
    OpenField(name='story', type='STRING'),
    OpenField(name='tags', type='STRING'),
    OpenField(name='year', type='STRING'),
    OpenField(name='actor_ids', type='STRING'),
    OpenField(name='director_ids', type='STRING'),
    OpenField(name='dt', type='STRING')
)

# Create the feature view (registers metadata only — data is not written yet)
feature_view_movie_name = "feature_view_movie"
batch_feature_view = project.get_feature_view(feature_view_movie_name)
if batch_feature_view is None:
    batch_feature_view = project.create_batch_feature_view(
        name=feature_view_movie_name,
        schema=movie_schema,
        online=True,
        entity=cur_entity_name_movie,
        primary_key='movie_id',
        partitions=['dt'],
        ttl=-1
    )

# Write data to the offline store
cur_task = batch_feature_view.write_table(ds, partitions={'dt': '20220830'})
cur_task.wait()
print(cur_task.task_summary)

# Publish to the online store
cur_task = batch_feature_view.publish_table({'dt': '20220830'})
cur_task.wait()
print(cur_task.task_summary)

# Verify
batch_feature_view = project.get_feature_view(feature_view_movie_name)
batch_feature_view.print_summary()

Vues de fonctionnalités User et Rating

Les tables user et rating suivent le même schéma : chargement de la source, définition du schéma, création de la vue de fonctionnalités, écriture et publication :

# User feature view
users_path = 'https://feature-store-test.oss-cn-beijing.aliyuncs.com/dataset/moviedata_all/users.csv'
ds = UrlDataSource(users_path, delimiter=',', omit_header=True)

user_schema = OpenSchema(
    OpenField(name='user_md5', type='STRING'),
    OpenField(name='user_nickname', type='STRING'),
    OpenField(name='ds', type='STRING')
)

feature_view_user_name = "feature_view_users"
batch_feature_view = project.get_feature_view(feature_view_user_name)
if batch_feature_view is None:
    batch_feature_view = project.create_batch_feature_view(
        name=feature_view_user_name,
        schema=user_schema,
        online=True,
        entity=cur_entity_name_user,
        primary_key='user_md5',
        partitions=['ds'],
        ttl=-1
    )

write_table_task = batch_feature_view.write_table(ds, {'ds': '20220830'})
write_table_task.wait()
print(write_table_task.task_summary)

cur_task = batch_feature_view.publish_table({'ds': '20220830'})
cur_task.wait()
print(cur_task.task_summary)

batch_feature_view = project.get_feature_view(feature_view_user_name)
batch_feature_view.print_summary()

# Rating feature view
ratings_path = 'https://feature-store-test.oss-cn-beijing.aliyuncs.com/dataset/moviedata_all/ratings.csv'
ds = UrlDataSource(ratings_path, delimiter=',', omit_header=True)

ratings_schema = OpenSchema(
    OpenField(name='rating_id', type='STRING'),
    OpenField(name='user_md5', type='STRING'),
    OpenField(name='movie_id', type='STRING'),
    OpenField(name='rating', type='STRING'),
    OpenField(name='rating_time', type='STRING'),
    OpenField(name='dt', type='STRING')
)

feature_view_rating_name = "feature_view_ratings"
batch_feature_view = project.get_feature_view(feature_view_rating_name)
if batch_feature_view is None:
    batch_feature_view = project.create_batch_feature_view(
        name=feature_view_rating_name,
        schema=ratings_schema,
        online=True,
        entity=cur_entity_name_ratings,
        primary_key='rating_id',
        event_time='rating_time',
        partitions=['dt']
    )

cur_task = batch_feature_view.write_table(ds, {'dt': '20220831'})
cur_task.wait()
print(cur_task.task_summary)

batch_feature_view = project.get_feature_view(feature_view_rating_name)
batch_feature_view.print_summary()

StreamFeatureView

StreamFeatureView gère les fonctionnalités en temps réel. Les données sont écrites directement dans le magasin en ligne.

Commencez par créer la table de données de test dans MaxCompute ou DataWorks :

CREATE TABLE IF NOT EXISTS online_stream_test_t1 (
  id STRING  COMMENT 'ID',
  count_value BIGINT  COMMENT 'Count value',
  metric_value DOUBLE COMMENT 'Metric value'
)
PARTITIONED BY (
    ds string COMMENT 'Data timestamp'
)
LIFECYCLE 365;

INSERT INTO TABLE online_stream_test_t1 PARTITION (ds='20250815')
SELECT
    CONCAT('str_', CAST(id AS STRING)) AS id,
    CAST(FLOOR(RAND() * 1000000) AS BIGINT) AS count_value,
    ROUND(RAND() * 1000, 2) AS metric_value
FROM (
SELECT SEQUENCE(1, 1000) AS id_list
) tmp
LATERAL VIEW EXPLODE(id_list) table_tmp AS id;

Une fois l'instruction SQL exécutée avec succès, la table online_stream_test_t1 est créée avec des données dans la partition ds=20250815.

Créez ensuite la StreamFeatureView et publiez-la :

online_schema = OpenSchema(
    OpenField(name='id', type='STRING'),
    OpenField(name='count_value', type='INT64'),
    OpenField(name='metric_value', type='DOUBLE')
)

feature_view_rating_name_stream = "feature_view_online_stream"
stream_feature_view = project.get_feature_view(feature_view_rating_name_stream)
if stream_feature_view is None:
    stream_feature_view = project.create_stream_feature_view(
        name=feature_view_rating_name_stream,
        schema=online_schema,
        online=True,
        entity=cur_entity_name_user,
        primary_key='id',
        event_time='count_value'
    )

stream_feature_view = project.get_feature_view(feature_view_rating_name_stream)
stream_feature_view.print_summary()

Dans une StreamFeatureView , le champ event_time sert à nettoyer les données expirées lorsqu'il est configuré. Pour plus de détails, consultez la section Cycle de vie des fonctionnalités en temps réel .

Synchronisez les données avec le magasin en ligne :

# Replace offline_datasource_id with your FeatureStore project's offline store ID.
# table_name is the offline feature table to push to the online store.
stream_task = stream_feature_view.publish_table(
    partitions={'ds': '20250815'},
    mode='Merge',
    offline_to_online=True,
    publish_config={
        'offline_datasource_id': project.offline_datasource_id,
        'table_name': 'online_stream_test_t1'
    }
)
stream_task.wait()
print(stream_task.task_summary)

Sequence FeatureView

Sequence FeatureView stocke les séquences de comportement utilisateur pour l'entraînement hors ligne et la récupération en ligne en temps réel.

Commencez par copier les données de séquence source depuis pai_online_project (accès en lecture public) vers votre propre projet :

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 l'instruction SQL exécutée avec succès, la table de fonctionnalités de séquence est créée avec les données des partitions ds=20231022, ds=20231023 et ds=20231024.

Créez la vue de fonctionnalités de séquence :

user_entity_name = "user"
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'  # Event time field in the behavior table
    item_id = 'item_id'             # Item ID field in the behavior table
    event = 'event'                 # Event type field in the behavior table

    # deduplication_method=1: deduplicates on ['user_id', 'item_id', 'event']
    # deduplication_method=2: deduplicates on ['user_id', 'item_id', 'event', 'event_time']
    sequence_feature_config_list = [
        SequenceFeatureConfig(
            offline_seq_name='click_seq_50_seq',  # Field name in the offline sequence table
            seq_event='click',                     # Event type to filter
            online_seq_name='click_seq_50',        # Name exposed to the online Go SDK
            seq_len=50                             # Maximum sequence length; longer sequences 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()

Synchronisez les données avec le magasin en ligne :

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

Enregistrez la table d'étiquettes pour la génération future d'ensembles d'entraînement :

label_table_name = 'fs_movie_feature_view_ratings_offline'
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='rating_time')

Étape 4 : Récupérer les fonctionnalités en ligne et définir des sélecteurs de fonctionnalités

Récupérer les fonctionnalités en ligne

Récupérez les fonctionnalités directement depuis une vue de fonctionnalités. FeatureStore privilégie actuellement FeatureDB pour la récupération des fonctionnalités en ligne.

feature_view_movie_name = "feature_view_movie"
batch_feature_view = project.get_feature_view(feature_view_movie_name)

# Retrieve features for a single item
ret_features_1 = batch_feature_view.list_feature_view_online_features(join_ids=['26357307'])
print("ret_features1 = ", ret_features_1)

# Retrieve features for multiple items in one call
ret_features_2 = batch_feature_view.list_feature_view_online_features(join_ids=['30444960', '3317352'])
print("ret_features2 = ", ret_features_2)

Définir un sélecteur de fonctionnalités

Un FeatureSelector spécifie les fonctionnalités à extraire d'une vue de fonctionnalités lors de la génération d'un ensemble d'entraînement ou de l'exécution d'une inférence. Trois modes de sélection sont pris en charge :

feature_view_name = 'feature_view_movie'

# Select specific features by name
feature_selector = FeatureSelector(feature_view_name, ['site_id', 'site_category'])

# Select all features
feature_selector = FeatureSelector(feature_view_name, '*')

# Select specific features and apply an alias
feature_selector = FeatureSelector(
    feature_view='user1',
    features=['f1', 'f2', 'f3'],
    alias={"f1": "f1_1"}  # Expose f1 as f1_1 in the output
)

Étape 5 : Créer et exporter un ensemble d'entraînement

Créer un ensemble d'entraînement

Un ensemble d'entraînement (table d'échantillons) combine une table d'étiquettes avec des fonctionnalités provenant d'une ou plusieurs vues de fonctionnalités, jointes sur les clés primaires à l'aide de jointures temporelles (point-in-time joins) pour éviter la fuite de données.

label_table_name = 'fs_movie_feature_view_ratings_offline'
output_ds = MaxComputeDataSource(data_source_id=project.offline_datasource_id)
train_set_output = TrainingSetOutput(output_ds)

# Select features from movie and user views
feature_movie_selector = FeatureSelector('feature_view_movie', ['name', 'actors', 'regions', 'tags'])
feature_user_selector = FeatureSelector('feature_view_users', ['user_nickname'])

train_set = project.create_training_set(
    label_table_name=label_table_name,
    train_set_output=train_set_output,
    feature_selectors=[feature_movie_selector, feature_user_selector]
)
print("train_set = ", train_set)

Enregistrer un modèle

Créez une entrée de modèle liée à l'ensemble d'entraînement :

model_name = "fs_rank_v1"
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)

Exporter l'ensemble d'entraînement

Spécifiez les partitions et l'heure de l'événement pour la table d'étiquettes et chaque vue de fonctionnalités, puis lancez l'exportation :

# Label table configuration
label_partitions = PartitionConfig(name='dt', value='20220831')
label_input_config = LabelInputConfig(
    partition_config=label_partitions,
    event_time='1999-01-00 00:00:00'
)

# Feature view configurations
movie_partitions = PartitionConfig(name='dt', value='20220830')
feature_view_movie_config = FeatureViewConfig(name='feature_view_movie', partition_config=movie_partitions)

user_partitions = PartitionConfig(name='ds', value='20220830')
feature_view_user_config = FeatureViewConfig(name='feature_view_users', partition_config=user_partitions)

feature_view_config_list = [feature_view_movie_config, feature_view_user_config]

# Output partition configuration
train_set_partitions = PartitionConfig(name='dt', value='20220831')
train_set_output_config = TrainSetOutputConfig(partition_config=train_set_partitions)

# Run the export
task = cur_model.export_train_set(label_input_config, feature_view_config_list, train_set_output_config)
task.wait()
print(task.summary)

Concepts clés

Magasin hors ligne

Le magasin hors ligne est un entrepôt de données destiné au stockage des fonctionnalités historiques. Les fonctionnalités sont écrites dans MaxCompute ou Hadoop Distributed File System (HDFS) à l'aide d'Apache Spark. Le magasin hors ligne remplit deux fonctions : générer des ensembles d'entraînement pour l'entraînement des modèles et fournir des fonctionnalités pour les prédictions par lots.

Magasin en ligne

Le magasin en ligne est un entrepôt de données dédié aux fonctionnalités en temps réel, offrant un accès à faible latence pour l'inférence en ligne. FeatureStore prend en charge FeatureDB, Hologres et Tablestore en tant que magasins en ligne.

Types de vues de fonctionnalités

Type	Description
`BatchFeatureView`	Fonctionnalités hors ligne ou J-1. Les données hors ligne sont écrites dans le magasin hors ligne et peuvent être publiées dans le magasin en ligne pour les requêtes en temps réel.
`StreamFeatureView`	Fonctionnalités en temps réel. Les données sont écrites directement dans le magasin en ligne.
`Sequence FeatureView`	Fonctionnalités de séquence de comportement utilisateur, prenant en charge les écritures hors ligne et les lectures en ligne en temps réel.

Étapes suivantes

Présentation de FeatureDB — découvrez le magasin en ligne qui assure la récupération des fonctionnalités en temps réel
Configurer les éléments FeatureStore — créez et gérez plusieurs projets
EasyRec sur GitHub — intégrez FeatureStore à EasyRec pour la génération de fonctionnalités (FG) et l'entraînement des modèles
FeatureGenerator sur GitHub — générez des fonctionnalités pour les modèles de recommandation

Pour toute question rencontrée lors de l'utilisation de FeatureStore, rejoignez le groupe de support DingTalk (ID de groupe : 32260796).