Platform For AI:Build recommendation systems with FeatureStore Python SDK

Sincronizar tabelas de dados

Um cenário de recomendação geralmente exige as seguintes tabelas de dados: uma tabela de features de usuário, uma tabela de itens, uma tabela de rótulos, uma tabela de features de sequência e uma tabela de comportamento.

Este tutorial prático utiliza tabelas de dados simuladas preparadas no projeto do MaxCompute pai_online_project. Para utilizá-las, execute comandos SQL no DataWorks para sincronizar as tabelas com seu próprio projeto do MaxCompute. Siga estas etapas:

1. Faça login no console do DataWorks.

2. No painel de navegação à esquerda, clique em Data Development & O&M > Data Development.

3. Selecione seu workspace do DataWorks e clique em Enter Data Development.

4. Passe o mouse sobre New e escolha New Node > MaxCompute > ODPS SQL. Configure os parâmetros do nó da seguinte forma:

   Parâmetro	Valor recomendado
   engine instance	Selecione o mecanismo do MaxCompute que você criou.
   node type	ODPS SQL
   path	Business Flow/Workflow/MaxCompute
   name	Insira um nome personalizado.

5. Clique em Confirm.

6. No editor de nós, execute os seguintes comandos SQL para sincronizar as tabelas de dados com seu projeto do MaxCompute. Em Resource Group, selecione seu grupo de recursos exclusivo.

   Tabela de usuários: 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';

   Após a conclusão do comando, a tabela conterá dados das seguintes partições:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Tabela de itens: 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';

   Após a conclusão do comando, a tabela conterá dados das seguintes partições:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Tabela de rótulos: 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'

   Após a conclusão do comando, a tabela conterá dados das seguintes partições:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Tabela de features de sequência: 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';

   Após a conclusão do comando, a tabela conterá dados das seguintes partições:

   • ds=20231022

   • ds=20231023

   • ds=20231024

   Tabela de comportamento: 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';

Agora é possível visualizar as tabelas sincronizadas no seu workspace. O restante deste tutorial utiliza essas cinco tabelas.

Configurar fontes de dados

O FeatureStore geralmente requer dois tipos de fontes de dados: uma fonte offline, como o MaxCompute, e uma fonte online, como FeatureDB, Hologres ou Tablestore. Este tutorial usa MaxCompute e FeatureDB como exemplos.

1. Faça login no PAI console. No painel de navegação à esquerda, escolha Data Preparation > FeatureStore.

2. Selecione um workspace e clique em Enter FeatureStore.

3. Configure uma fonte de dados do MaxCompute.

   1. Na aba Create Store, clique em Create Store e, na página exibida, configure os parâmetros da fonte de dados do MaxCompute.

      Parâmetro	Valor recomendado
      Type	MaxCompute
      Name	Insira um nome personalizado.
      MaxCompute Project Name	Selecione o projeto do MaxCompute que você criou.

   2. Clique em Submit.

4. Configure uma fonte de dados do FeatureDB.

   1. Caso já tenha criado uma fonte de dados do FeatureDB, pule esta etapa.

   2. Na aba Store, clique em Create Store e configure os parâmetros da fonte de dados do FeatureDB na caixa de diálogo.

      Parâmetro	Valor recomendado
      Type	FeatureDB (Se for sua primeira vez usando o FeatureDB, siga as instruções na tela para ativá-lo.)
      Name	Nomes personalizados não são suportados. O nome padrão é feature_db.
      Username	Defina um nome de usuário.
      Password	Defina uma senha.
      VPC network high-speed connection (optional)	Esta opção permite usar o FeatureStore SDK para acessar diretamente o FeatureDB de dentro de uma VPC via rede privada. Essa conexão melhora o desempenho de leitura/gravação de dados e reduz a latência de acesso.
      VPC	Selecione a VPC que contém seus serviços online do FeatureStore.
      Zone and vSwitch	Selecione uma zona e um vSwitch. Certifique-se de que o vSwitch esteja na mesma zona dos seus serviços online. Para alta disponibilidade, recomendamos selecionar vSwitches em pelo menos duas zonas.

   3. Clique em 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))

Projeto do FeatureStore

É possível criar vários projetos no FeatureStore, onde cada projeto funciona como um namespace isolado. Para instruções detalhadas, consulte Configure a FeatureStore project. A execução deste notebook requer um serviço de backend do FeatureStore e fontes de dados configuradas. Após ativar o FeatureStore, configure uma fonte de dados. Para mais informações, consulte Create a data source.

O parâmetro offline_datasource_id especifica o ID da fonte de dados offline, e online_datasource_id especifica o ID da fonte de dados online.

Este exemplo utiliza um projeto chamado 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.")

Execute o código a seguir para recuperar o projeto atual e imprimir suas informações.

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

Entidade de Feature

Uma entidade de feature descreve uma coleção de features relacionadas. Várias visualizações de features podem ser associadas a uma única entidade. Cada entidade possui um join_id, que une features de diferentes visualizações. Cada visualização de feature tem uma chave primária para recuperar seus dados, e o nome dessa chave pode ser diferente do join_id.

Em sistemas de recomendação, as features geralmente estão associadas a uma entidade de usuário ou de item. Este exemplo demonstra como criar tanto uma entidade user quanto uma entidade item.

• Criar entidade de usuário

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

• Criar entidade de item

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

Visualização de Feature

Uma visualização de feature é um objeto central no FeatureStore que ingere dados externos. Ela define a origem dos dados (fonte de dados), quaisquer operações de pré-processamento ou transformação, a estrutura dos dados (schema da feature, incluindo nomes e tipos) e o local de armazenamento (armazenamento online e/ou offline). A visualização também gerencia metadados, como chave primária, tempo do evento, chave de partição, entidade de feature associada e TTL (Time to Live). Um TTL padrão de -1 indica que a feature é permanentemente válida. Um valor positivo de TTL significa que consultas online recuperarão apenas os dados de feature mais recentes dentro da janela de tempo especificada.

Existem três tipos de visualizações de feature: Batch FeatureView (para features offline ou T-1), Stream FeatureView (para features em tempo real) e Sequence FeatureView (para features de sequência).

Batch FeatureView

Uma Batch FeatureView injeta dados offline no armazenamento offline do FeatureStore e pode, opcionalmente, sincronizá-los com o armazenamento online para suportar consultas em tempo real. Utilize esta visualização para features offline ou T-1.

• Registre a tabela de features offline de usuário

  1. Registre a tabela rec_sln_demo_user_table_preprocess_all_feature_v1 no 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. Sincronize os dados da partição 20231023 da tabela rec_sln_demo_user_table_preprocess_all_feature_v1 da fonte offline para a fonte online.

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

  3. Verifique o status da tarefa.

     user_task.print_summary()

• Registre a tabela de features offline de item

  1. Registre a tabela rec_sln_demo_item_table_preprocess_all_feature_v1 no 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. Sincronize os dados da partição 20231023 da tabela rec_sln_demo_item_table_preprocess_all_feature_v1 da fonte offline para a fonte online.

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

  3. Verifique o status da tarefa.

     item_task.print_summary()

Sequence FeatureView

Uma Sequence FeatureView suporta a escrita de features de sequência offline e a leitura de features de sequência em tempo real online. Em cenários de recomendação, uma tabela de features de sequência offline (F1) é inicialmente gerada por simulação e posteriormente substituída por logs online. Durante consultas online para sequências em tempo real, os dados são recuperados de duas tabelas de comportamento online: uma tabela de comportamento T-1 (B1) e uma tabela de comportamento em tempo real do dia atual (dia T) (B2). A tabela B2 contém features atualizadas por computação em tempo real. O FeatureStore consulta B1 e B2, constrói a sequência de features do usuário e a combina com outras features para pontuação do modelo.

A tabela de comportamento online T-1 (B1) é normalmente sincronizada a partir de uma tabela de comportamento offline T-1 (A1), com o FeatureStore lidando automaticamente com operações como deduplicação. A tabela de comportamento online do dia atual (dia T) (B2) atualmente exige que você escreva dados através de uma API ou outros produtos da Alibaba Cloud, como o Flink.

Portanto, ao registrar uma Sequence FeatureView, o FeatureStore gerencia quatro tabelas em seu nome: a tabela de sequência offline (F1), a tabela de comportamento offline T-1 (A1), a tabela de comportamento online T-1 (B1) e a tabela de comportamento online do dia T (B2).

Durante o registro, você fornece apenas a tabela de sequência offline (F1) e a tabela de comportamento offline T-1 (A1). O FeatureStore cria as tabelas online e cuida da sincronização e deduplicação.

1. Registre a 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. Sincronize os dados da partição 20231023 da tabela rec_sln_demo_behavior_table_preprocess_v3 da fonte offline para a fonte online. Durante a sincronização, o sistema verifica automaticamente se há dados dos N dias anteriores e os preenche se estiverem ausentes. Você pode especificar N usando o parâmetro days_to_load, cujo padrão é 30. Recomendamos usar o valor padrão na maioria dos casos.

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

3. Verifique o status da tarefa.

   seq_task.print_summary()

Stream FeatureView

Uma Stream FeatureView grava dados diretamente no armazenamento online e os sincroniza simultaneamente com o armazenamento offline. Utilize-a para cenários que exigem atualizações de features em tempo real, como atualização de preços de itens ou volumes de vendas.

Registrar a tabela de rótulos

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)

Recuperar features online

É possível recuperar features do armazenamento online para finalidades como depuração de consistência offline-online e análise de dados. Este recurso é atualmente otimizado para 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)

Conjunto de treinamento

Para treinar um modelo, primeiro construa um conjunto de treinamento com dados de rótulos e features. Ao interagir com o FeatureStore, você fornece os dados de rótulo, especifica as features a serem recuperadas e o sistema realiza uma junção ponto-no-tempo baseada na chave primária (se existir um tempo de evento).

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

Features do modelo

Após treinar um modelo e implantá-lo como serviço, você pode usá-lo para previsões de negócios. As amostras de treinamento estão disponíveis no objeto train_set da etapa anterior.

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)