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

Veri tablolarını senkronize edin

Tipik bir öneri senaryosu üç veri tablosu gerektirir: kullanıcı özellikleri, öğe özellikleri ve etiketler. Bu öğreticide pai_online_project MaxCompute projesindeki simüle edilmiş tablolar kullanılır.

Bu tabloları pai_online_project konumundan kendi MaxCompute projenize senkronize etmek için DataWorks'te aşağıdaki SQL kodunu çalıştırın.

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

Veri kaynaklarını yapılandırın

FeatureStore, çevrimdışı bir depo (MaxCompute) ve çevrimiçi bir depo (FeatureDB, Hologres veya Tablestore) gerektirir. Bu öğreticide MaxCompute ve FeatureDB kullanılmaktadır.

1. PAI konsolunda oturum açın. Gezinme bölmesinden Data Preparation > FeatureStore seçeneklerine tıklayın.

2. Bir çalışma alanı seçin ve Enter FeatureStore seçeneğine tıklayın.

3. MaxCompute veri kaynağını yapılandırın. Data Source sekmesinde Create Store seçeneğine tıklayın. Aşağıdaki parametreleri ayarlayın.

   Parametre	Önerilen değer
   Type	MaxCompute
   Name	Özel bir ad girin.
   MaxCompute Project Name	Oluşturduğunuz MaxCompute projesini seçin.

4. FeatureDB veri kaynağını yapılandırın. Daha önce oluşturduysanız bu adımdan geçin. Store sekmesinde Create Store seçeneğine tıklayın. Aşağıdaki parametreleri ayarlayın. Submit seçeneğine tıklayın.

   Parametre	Önerilen değer
   Type	FeatureDB (İlk kez kullanıyorsanız etkinleştirmek için ekrandaki talimatları izleyin)
   Name	Özelleştirilemez. Varsayılan değer feature_db'dir.
   Username	Bir kullanıcı adı belirleyin.
   Password	Bir parola belirleyin.
   VPC high-speed connection (Optional)	Yapılandırma başarıyla tamamlandıktan sonra FeatureStore SDK'sını FeatureStore SDK bir VPC içinde kullanarak PrivateLink aracılığıyla FeatureDB'ye erişin. Bu işlem gecikme süresini azaltır ve performansı artırır.
   VPC	Çevrimiçi FeatureStore hizmetinizin bulunduğu VPC'yi seçin.
   Zone and vSwitch	Bir alan ve vSwitch seçin. Çevrimiçi hizmet makinenizin bulunduğu alana ait vSwitch'i seçin. Yüksek kullanılabilirlik için en az iki alandan vSwitch'ler seçin.

FeatureStore Python SDK'yı yükleyin

SDK'yı DataWorks zamanlama için özel kaynak grubuna yükleyin.

1. DataWorks konsolunda oturum açın.

2. Sol gezinme bölmesinden Resource Group seçeneğine tıklayın.

3. Exclusive Resource Groups sekmesinde Purpose değeri Data Scheduling olarak ayarlanmış kaynak grubunu bulun. Zamanlama kaynağı için simgesine tıklayın ve O&M Assistant seçeneğini belirleyin.

4. Create Command seçeneğine tıklayın. Aşağıdaki parametreleri ayarlayın.

   Parametre	Önerilen değer
   Command Name	Özel bir ad girin. Bu öğreticide install kullanılmaktadır.
   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	Özel bir süre girin.

5. Komutu oluşturmak için OK seçeneğine tıklayın.

6. Command seçeneğine tıklayın. Görüntülenen sayfada Running seçeneğine tıklayın.

7. En son yürütme durumunu görüntülemek için düğmesine tıklayın. Durum Successful olarak değiştiğinde yükleme tamamlanmış demektir.

Yöntem 1: Konsolu kullanın

Yöntem 2: Python SDK'yı kullanın

SDK kullanım adımları için DSW Gallery adresine bakın.

Eğitim veri kümesini dışa aktarın

1. DataWorks konsolunda oturum açın.

2. Sol gezinme bölmesinden Data Development & O&M > Data Development seçeneklerine tıklayın.

3. Oluşturduğunuz DataWorks çalışma alanını seçin ve Go to Data Studio seçeneğine tıklayın.

4. Create üzerine gelin ve Create Node > MaxCompute > PyODPS 3 yolunu izleyin. Aşağıdaki parametreleri ayarlayın.

   Parametre	Önerilen değer
   Engine Instance	Oluşturduğunuz MaxCompute motorunu seçin.
   Node Type	PyODPS 3
   Path	Business Flow/Workflow/MaxCompute
   Name	Özel bir ad girin.

5. Confirm seçeneğine tıklayın.

6. Aşağıdaki betiği düğüme kopyalayı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. Sağ bölmede Scheduling Configuration seçeneğine tıklayın. Aşağıdaki parametreleri ayarlayın.

   Parametre	Önerilen değer
   Scheduling Parameters
   Parameter Name	dt
   Parameter Value	$[yyyymmdd-1]
   Resource Properties
   Scheduling Resource Group	Oluşturduğunuz zamanlama için özel kaynak grubunu seçin.
   Scheduling Dependencies	Oluşturduğunuz kullanıcı ve öğe tablolarını seçin.

2. Düğümü yapılandırıp test ettikten sonra düğüm yapılandırmasını kaydedin ve gönderin.

3. Veri geri doldurma işlemini gerçekleştirin.

Dışa aktarma görevini görüntüleyin (isteğe bağlı)

1. FeatureStore Projects sayfasında bir proje adına tıklayarak ayrıntılar sayfasını açın.

2. Feature Entity sekmesinde Jobs seçeneğine tıklayın.

3. Hedef görevin satırında Details seçeneğine tıklayarak temel bilgileri, çalıştırma yapılandırmasını ve görev günlüklerini görüntüleyin.

Modeli eğitin

EasyRec, FeatureStore ile entegre çalışan açık kaynaklı bir öneri çerçevesidir; modelleri eğitmek, dışa aktarmak ve yayınlamak için kullanılır. EasyRec ile model eğitmek için girdi olarak fs_demo_fs_rank_v1_trainning_set tablosunu kullanın.

EasyRec ile ilgili sorularınız için DingTalk üzerindeki PAI danışma grubuna katılın (Grup Kimliği: 32260796).

Veri senkronizasyonunu zamanlayın

Kullanıcı ve öğe özellik tablolarını çevrimdışı depodan çevrimiçi depoya düzenli olarak senkronize edecek şekilde zamanlayın.

1. DataWorks konsolunda oturum açın.

2. Sol gezinme bölmesinden Data Development & O&M > Data Development seçeneklerine tıklayın.

3. Oluşturduğunuz DataWorks çalışma alanını seçin ve Go to Data Studio seçeneğine tıklayın.

4. user_table_preprocess_all_feature_v1 için senkronizasyonu zamanlayın (Ayrıntıları görüntülemek için tıklayın)

   1. New üzerine gelin ve New Node > MaxCompute > PyODPS 3 yolunu izleyin.

   2. user_table_preprocess_all_feature_v1 için senkronizasyonu zamanlamak üzere aşağıdaki betiği kopyalayın. ``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. Sağ bölmede Scheduling Configuration seçeneğine tıklayın. Aşağıdaki parametreleri ayarlayın. | Parametre | Önerilen değer | |---|---| | Scheduling Parameters | | | Parameter Name | dt | | Parameter Value | $[yyyymmdd-1] | | Resource Properties | | | Scheduling Resource Group | Oluşturduğunuz zamanlama için özel kaynak grubunu seçin. | | Scheduling Dependencies | Oluşturduğunuz kullanıcı tablosunu seçin. |

   4. Düğümü yapılandırıp test ettikten sonra düğüm yapılandırmasını kaydedin ve gönderin.

   5. Veri geri doldurma işlemini gerçekleştirin.

5. item_table_preprocess_all_feature_v1 için senkronizasyonu zamanlayın (Ayrıntıları görüntülemek için tıklayın)

   1. Create üzerine gelin ve Create Node > MaxCompute > PyODPS 3 yolunu izleyin. Düğüm parametrelerini yapılandırın.

   2. Confirm seçeneğine tıklayın.

   3. Aşağıdaki betiği kopyalayın. ``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. Sağ bölmede Scheduling Configuration seçeneğine tıklayın. Aşağıdaki parametreleri ayarlayın. | Parametre | Önerilen değer | |---|---| | Scheduling Parameters | | | Parameter Name | dt | | Parameter Value | $[yyyymmdd-1] | | Resource Properties | | | Scheduling Resource Group | Oluşturduğunuz zamanlama için özel kaynak grubunu seçin. | | Scheduling Dependencies | Oluşturduğunuz öğe tablosunu seçin. |

   5. Düğümü yapılandırıp test ettikten sonra düğüm yapılandırmasını kaydedin ve gönderin.

   6. Veri geri doldurma işlemini gerçekleştirin.

EAS model hizmeti oluşturun ve dağıtın

EasyRec İşlemcisi, düşük gecikme süreli ve yüksek performanslı özellik alma işlemleri için FeatureStore C++ SDK'sını içerir. Öneri motorundan puanlama isteklerini alır, özellikleri getirir, çıkarım çalıştırır ve puanları döndürür.

1. DataWorks konsolunda oturum açın.

2. Sol gezinme bölmesinden Data Development & O&M > Data Development seçeneklerine tıklayın.

3. Oluşturduğunuz DataWorks çalışma alanını seçin ve Go to Data Studio seçeneğine tıklayın.

4. New üzerine gelin ve New Node > MaxCompute > PyODPS 3 yolunu izleyin.

5. Aşağıdaki betiği kopyalayın.

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. Sağ bölmede Scheduling Configuration seçeneğine tıklayın. Aşağıdaki parametreleri ayarlayın.

   Parametre	Önerilen değer
   Scheduling Parameters
   Parameter Name	dt
   Parameter Value	$[yyyymmdd-1]
   Resource Properties
   Scheduling Resource Group	Oluşturduğunuz zamanlama için özel kaynak grubunu seçin.
   Scheduling Dependencies	İlgili eğitim görevini ve item_table_preprocess_all_feature_v1'i seçin.

2. Düğümü yapılandırıp test ettikten sonra çalıştırın ve dağıtım durumunu kontrol edin.

3. Dağıtım tamamlandıktan sonra 34. satırı yorum haline getirin, 37. satırın yorumunu kaldırın ve görevi zamanlanmış yürütme için gönderin. > İpucu: İlk dağıtım için 34. satır create komutunu kullanır. Sonraki zamanlanmış güncellemeler için 37. satır modify komutunu kullanır. İlk dağıtım başarılı olduktan sonra modify komutuna geçin.

4. (İsteğe bağlı) Dağıtılan hizmeti Elastic Algorithm Service (EAS) sayfasının Inference Service sekmesinde görüntüleyin. Özel dağıtım.

5. (İsteğe bağlı) Veri kaynağınız VPC erişimi gerektiriyorsa (Hologres gibi) EAS ve veri kaynağı VPC'lerini bağlayın. Hologres için:

   1. Hologres örneğinin Network Information sayfasında VPC ID ve vSwitch ID değerlerini bulun.

   2. EAS hizmeti sayfasında sağ üst köşede Configure High-Speed Connection seçeneğine tıklayın.

   3. İlgili VPC ID ve vSwitch ID değerlerini girin.

   4. Bir Security Group Name girin. Mevcut bir güvenlik grubunu seçin veya yeni bir tane oluşturun. Hologres genellikle 80 numaralı bağlantı noktasını kullanır. Güvenlik grubunun bu bağlantı noktasındaki trafiğe izin vermesi gerekir.

   5. OK seçeneğine tıklayın. Güncelleme tamamlandıktan sonra hizmet kullanılabilir hale gelir.

PAI-REC'i yapılandırın

PAI-REC, FeatureStore Go SDK'sını entegre eden bir öneri motoru hizmetidir. FeatureStore ve EAS ile bağlantı kurar.

"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"
            }
        }
    ],