Platform For AI:FeatureStore LLM embedding best practices

Step 1: Create a DataHub topic

1. Log on to the DataHub console and create a project.

2. Click the project name to open the project details page.

3. Click Create Topic and configure the following parameters. Keep other parameters at their default values.

   Parameter	Description
   Name	Enter a custom name.
   Schema Details	Add two fields:
   	- item_id, BIGINT, clear the Allow Null checkbox.
   	- review_text, STRING, select the Allow NULL checkbox.

4. Click Create.

5. Click the created topic to open the topic details page.

6. On the Subscription List tab, click Subscribe, fill in the relevant information, and click Create.

Step 2: Write Flink SQL to bridge DataHub and FeatureStore

The Flink SQL job has three parts: a source table reading from DataHub, a sink table writing to the FeatureStore Connector, and an INSERT statement linking the two.

CREATE TEMPORARY TABLE item_fea_embedding_debug_v1_dh
(
    item_id      bigint
    ,review_text string
)
WITH (
    'connector' = 'datahub'
    ,'subId' = '1747297545688WFPX0'
    ,'endPoint' = 'http://dh-cn-beijing.aliyuncs.com'
    ,'project' = 'fs_test'
    ,'topic' = 'feature_store_llm_embedding_test_v1'
    ,'accessId' = 'xxx'
    ,'accessKey' = 'xxx'
)
;

CREATE TEMPORARY TABLE item_fea_embedding_debug_v1
(
    item_id          bigint
    ,review_text         string
)
WITH (
    'connector' = 'featurestore'
    ,'region_id' = 'cn-beijing'
    ,'project' = 'fs_demo_featuredb'
    ,'feature_view' = 'item_fea_embedding_debug_v1'
    ,'username' = 'xxx'  -- Replace with your FeatureDB username
    ,'password' = 'xxx'  -- Replace with your FeatureDB password
    ,'aliyun_access_id' = 'xxx'
    ,'aliyun_access_key' = 'xxx'
)
;

INSERT INTO item_fea_embedding_debug_v1
SELECT
    item_id
    ,review_text
FROM item_fea_embedding_debug_v1_dh
;

The FeatureStore Connector automatically invokes the registered LLM to extract embeddings before writing to FeatureDB.

Step 3: Verify online features

Use the FeatureStore SDK to read features from the online store and confirm the real-time pipeline works.

Full sync

For initial deployments, synchronize the entire offline table to FeatureDB. Copy the following code to a PyODPS 3 node in DataWorks and run it.

from feature_store_py.fs_client import FeatureStoreClient
import datetime
from feature_store_py.fs_datasource import MaxComputeDataSource
import sys
from odps.accounts import StsAccount

cur_day = args['dt']
print('cur_day = ', cur_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_featuredb'
project = fs.get_project(cur_project_name)

feature_view_name = 'item_fea_embedding_debug_v1'
batch_feature_view = project.get_feature_view(feature_view_name)
task = batch_feature_view.publish_table(partitions={'ds':cur_day}, mode='Merge', offline_to_online=True, publish_config={'offline_datasource_id':19, 'table_name': 'item_fea_review_text_v1'})
task.wait()

The publish_table() method triggers LLM embedding extraction and writes the results to FeatureDB.

Incremental sync

After the initial full sync, synchronize only new records on a daily schedule. This involves two steps: computing the delta and publishing it.

Step 1: Compute the new item set. Run this SQL in DataWorks to identify items added since the previous day.

CREATE TABLE IF NOT EXISTS item_fea_review_text_new_tmp_v1 (
  item_id STRING COMMENT 'Item ID',
  review_text STRING COMMENT 'English description'
)
PARTITIONED BY (
    ds string COMMENT 'data timestamp'
)
LIFECYCLE 365
;

-- Insert new data
INSERT INTO TABLE item_fea_review_text_new_tmp_v1 PARTITION (ds='${bdp.system.bizdate}')
SELECT
    t1.item_id,
    t1.review_text
FROM
    item_fea_review_text_v1 t1
WHERE
    t1.ds = '${bdp.system.bizdate}'
    AND NOT EXISTS (
      SELECT 1
      FROM item_fea_review_text_v1 t2
      WHERE t2.ds = TO_CHAR(DATEADD(TO_DATE('${bdp.system.bizdate}','yyyymmdd'), -1, 'dd'),'yyyymmdd')
      AND t1.item_id = t2.item_id
    );

Step 2: Publish the incremental table. Copy the following code to a PyODPS 3 node in DataWorks and run it. The only difference from the full sync script is the table name in publish_config.

from feature_store_py.fs_client import FeatureStoreClient
import datetime
from feature_store_py.fs_datasource import MaxComputeDataSource
import sys
from odps.accounts import StsAccount

cur_day = args['dt']
print('cur_day = ', cur_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_featuredb'
project = fs.get_project(cur_project_name)

feature_view_name = 'item_fea_embedding_debug_v1'
batch_feature_view = project.get_feature_view(feature_view_name)
task = batch_feature_view.publish_table(partitions={'ds':cur_day}, mode='Merge', offline_to_online=True, publish_config={'offline_datasource_id':19, 'table_name': 'item_fea_review_text_new_tmp_v1'})
task.wait()