Platform For AI:Manage features in a recommendation system by using FeatureStore

Sincronizar tabelas de dados

Cenários típicos de recomendação exigem três tabelas de dados: recursos de usuário, recursos de item e rótulos. Este tutorial utiliza tabelas simuladas no projeto do MaxCompute pai_online_project.

Execute o SQL a seguir no DataWorks para sincronizar essas tabelas de pai_online_project para seu projeto do MaxCompute.

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}';

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}';

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}';

Configurar fontes de dados

O FeatureStore requer um armazenamento offline (MaxCompute) e um armazenamento online (FeatureDB, Hologres ou Tablestore). Este tutorial usa MaxCompute e FeatureDB.

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

2. Selecione um workspace e clique em Enter FeatureStore.

3. Configure a fonte de dados do MaxCompute. Na aba Data Source, clique em Create Store e defina os seguintes parâmetros.

   Parâmetro	Valor sugerido
   Type	MaxCompute
   Name	Insira um nome personalizado.
   MaxCompute Project Name	Selecione o projeto do MaxCompute criado.

4. Configure a fonte de dados do FeatureDB. Caso já tenha criado uma, pule esta etapa. Na aba Store, clique em Create Store, defina os parâmetros a seguir e clique em Submit.

   Parâmetro	Valor sugerido
   Type	FeatureDB (Se for o primeiro uso, siga as instruções na tela para ativar)
   Name	Não personalizável. O valor padrão é feature_db.
   Username	Defina um nome de usuário.
   Password	Defina uma senha.
   VPC high-speed connection (Optional)	Após configurar com êxito, use o SDK do FeatureStore em uma VPC para acessar o FeatureDB via PrivateLink. Isso reduz a latência e melhora o desempenho.
   VPC	Selecione a VPC onde está localizado seu serviço online do FeatureStore.
   Zone and vSwitch	Selecione uma zona e um vSwitch. Escolha o vSwitch da zona onde está localizada sua máquina de serviço online. Selecione vSwitches em pelo menos duas zonas para garantir alta disponibilidade.

Instalar o SDK do FeatureStore para Python

Instale o SDK no grupo de recursos exclusivo do DataWorks para agendamento.

1. Faça login no console do DataWorks.

2. No painel de navegação à esquerda, clique em Resource Group.

3. Na aba Exclusive Resource Groups, localize o grupo de recursos onde Purpose está definido como Data Scheduling. Clique no ícone do recurso de agendamento e escolha O&M Assistant.

4. Clique em Create Command e defina os seguintes parâmetros.

   Parâmetro	Valor sugerido
   Command Name	Insira um nome personalizado. Este tutorial usa install.
   Command Type	Manual Input (pip Command Cannot Be Used To Install Third-party Packages)
   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-2.0.2-py3-none-any.whl
   Timeout	Insira um tempo personalizado.

5. Clique em OK para criar o comando.

6. Clique em Command e, na página exibida, clique em Running.

7. Clique no botão para visualizar o status de execução mais recente. Quando o status mudar para Successful, a instalação estará concluída.

Método 1: Usar o console

Método 2: Usar o SDK em Python

Para consultar as etapas de uso do SDK, acesse a DSW Gallery.

Exportar o conjunto de treinamento

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 o workspace do DataWorks criado e clique em Go to Data Studio.

4. Passe o mouse sobre Create e escolha Create Node > MaxCompute > PyODPS 3. Defina os seguintes parâmetros.

   Parâmetro	Valor sugerido
   Engine Instance	Selecione o mecanismo do MaxCompute criado.
   Node Type	PyODPS 3
   Path	Business Flow/Workflow/MaxCompute
   Name	Insira um nome personalizado.

5. Clique em Confirm.

6. Copie o script a seguir para o nó.

from feature_store_py.fs_client import FeatureStoreClient
from feature_store_py.fs_project import FeatureStoreProject
from feature_store_py.fs_datasource import LabelInput, MaxComputeDataSource, TrainingSetOutput
from feature_store_py.fs_features import FeatureSelector
from feature_store_py.fs_config import LabelInputConfig, PartitionConfig, FeatureViewConfig
from feature_store_py.fs_config import TrainSetOutputConfig, EASDeployConfig
import datetime
import sys
from odps.accounts import StsAccount

cur_day = args['dt']
print('cur_day = ', cur_day)
offset = datetime.timedelta(days=-1)
pre_day = (datetime.datetime.strptime(cur_day, "%Y%m%d") + offset).strftime('%Y%m%d')
print('pre_day = ', pre_day)

access_key_id = o.account.access_id
access_key_secret = o.account.secret_access_key
sts_token = None
endpoint = 'paifeaturestore-vpc.cn-beijing.aliyuncs.com'
if isinstance(o.account, StsAccount):
    sts_token = o.account.sts_token
fs = FeatureStoreClient(access_key_id=access_key_id, access_key_secret=access_key_secret, security_token=sts_token, endpoint=endpoint)
cur_project_name = 'fs_demo'
project = fs.get_project(cur_project_name)

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)
feature_view_config_list = [feature_view_user_config, feature_view_item_config]
train_set_partitions = PartitionConfig(name = 'ds', value = cur_day)
train_set_output_config = TrainSetOutputConfig(partition_config=train_set_partitions)

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

1. No painel à direita, clique em Scheduling Configuration e defina os seguintes parâmetros.

   Parâmetro	Valor sugerido
   Scheduling Parameters
   Parameter Name	dt
   Parameter Value	$[yyyymmdd-1]
   Resource Properties
   Scheduling Resource Group	Selecione o grupo de recursos exclusivo para agendamento criado.
   Scheduling Dependencies	Selecione a tabela de usuários e a tabela de itens criadas.

2. Após configurar e testar o nó, salve e envie a configuração.

3. Realize o backfill de dados.

Visualizar tarefa de exportação (opcional)

1. Na página Projects do FeatureStore, clique no nome de um projeto para abrir sua página de detalhes.

2. Na aba Feature Entity, clique em Jobs.

3. Clique em Details na linha da tarefa desejada para visualizar informações básicas, configuração de execução e logs.

Treinar o modelo

O EasyRec é um framework de recomendação open-source que se integra ao FeatureStore para treinar, exportar e publicar modelos. Utilize a tabela fs_demo_fs_rank_v1_trainning_set como entrada para treinar um modelo com o EasyRec.

Para dúvidas sobre o EasyRec, participe do grupo de consultoria do PAI no DingTalk (ID do Grupo: 32260796).

Agendar sincronização de dados

Agende a sincronização regular das tabelas de recursos de usuários e itens do armazenamento offline para o online.

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 o workspace do DataWorks criado e clique em Go to Data Studio.

4. Agendar sincronização para user_table_preprocess_all_feature_v1 (Clique para ver detalhes)

   1. Passe o mouse sobre New e escolha New Node > MaxCompute > PyODPS 3.

   2. Copie o script a seguir para agendar a sincronização de user_table_preprocess_all_feature_v1. ``python 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. No painel à direita, clique em Scheduling Configuration e defina os seguintes parâmetros. | Parâmetro | Valor sugerido | |---|---| | Scheduling Parameters | | | Parameter Name | dt | | Parameter Value | $[yyyymmdd-1] | | Resource Properties | | | Scheduling Resource Group | Selecione o grupo de recursos exclusivo para agendamento criado. | | Scheduling Dependencies | Selecione a tabela de usuários criada. |

   4. Após configurar e testar o nó, salve e envie a configuração.

   5. Realize o backfill de dados.

5. Agendar sincronização para item_table_preprocess_all_feature_v1 (Clique para ver detalhes)

   1. Passe o mouse sobre Create e escolha Create Node > MaxCompute > PyODPS 3. Configure os parâmetros do nó.

   2. Clique em Confirm.

   3. Copie o script a seguir. ``python 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. No painel à direita, clique em Scheduling Configuration e defina os seguintes parâmetros. | Parâmetro | Valor sugerido | |---|---| | Scheduling Parameters | | | Parameter Name | dt | | Parameter Value | $[yyyymmdd-1] | | Resource Properties | | | Scheduling Resource Group | Selecione o grupo de recursos exclusivo para agendamento criado. | | Scheduling Dependencies | Selecione a tabela de itens criada. |

   5. Após configurar e testar o nó, salve e envie a configuração.

   6. Realize o backfill de dados.

Criar e implantar o serviço de modelo EAS

O EasyRec Processor inclui o SDK do FeatureStore em C++ para recuperação de recursos com baixa latência e alto desempenho. Ele recebe solicitações de pontuação do mecanismo de recomendação, busca os recursos, executa a inferência e retorna as pontuações.

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 o workspace do DataWorks criado e clique em Go to Data Studio.

4. Passe o mouse sobre New e escolha New Node > MaxCompute > PyODPS 3.

5. Copie o script a seguir.

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",  # Path of the trained model. customize the path.
  "model_config": {
    "access_key_id": f'{o.account.access_id}',
    "access_key_secret": f'{o.account.secret_access_key}',
    "region": "cn-beijing",  # Replace this with the region where PAI is deployed. This tutorial uses cn-beijing as an example.
    "fs_project": "fs_demo",  # Replace this with the name of your FeatureStore project. This tutorial uses fs_demo as an example.
    "fs_model": "fs_rank_v1",  # Replace this with the name of your FeatureStore model feature. This tutorial uses fs_rank_v1 as an example.
    "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 line for the first deployment
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 scheduled 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")

1. No painel à direita, clique em Scheduling Configuration e defina os seguintes parâmetros.

   Parâmetro	Valor sugerido
   Scheduling Parameters
   Parameter Name	dt
   Parameter Value	$[yyyymmdd-1]
   Resource Properties
   Scheduling Resource Group	Selecione o grupo de recursos exclusivo para agendamento criado.
   Scheduling Dependencies	Selecione a tarefa de treinamento correspondente e item_table_preprocess_all_feature_v1.

2. Após configurar e testar o nó, execute-o e verifique o status da implantação.

3. Quando a implantação for concluída, comente a linha 34, descomente a linha 37 e envie a tarefa para execução agendada. > Dica: A linha 34 usa create para a primeira implantação. A linha 37 usa modify para atualizações agendadas subsequentes. Após o sucesso da implantação inicial, mude para o comando modify.

4. (Opcional) Visualize o serviço implantado na aba Inference Service da página Elastic Algorithm Service (EAS). Implantação personalizada.

5. (Opcional) Se sua fonte de dados exigir acesso via VPC (como o Hologres), conecte as VPCs do EAS e da fonte de dados. Para o Hologres:

   1. Localize o VPC ID e o vSwitch ID na página Network Information da instância do Hologres.

   2. Na página do serviço EAS, clique em Configure High-Speed Connection no canto superior direito.

   3. Insira o VPC ID e o vSwitch ID correspondentes.

   4. Insira um Security Group Name. Selecione um grupo de segurança existente ou crie um novo. O Hologres geralmente usa a porta 80. O grupo de segurança deve permitir tráfego nesta porta.

   5. Clique em OK. O serviço ficará disponível após a conclusão da atualização.

Configurar o PAI-REC

O PAI-REC é um serviço de mecanismo de recomendação que integra o SDK do FeatureStore em Go. Ele se conecta ao FeatureStore e ao EAS.

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

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

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