Platform For AI:EasyRec

Background information

The following figure shows the architecture of a recommendation engine based on the EasyRec processor.

The EasyRec processor includes the following modules:

• Item Feature Cache: This module caches features from FeatureStore into memory, which reduces the burden on FeatureStore resulting from frequent request operations. Item Feature Cache supports incremental updates such as real-time feature updates.

• Feature Generator (FG): This module uses the same implementation for real-time and offline feature engineering to ensure consistency. FG is designed based on the extensive experience from Taobao.

• TFModel: This module uses TensorFlow to load the SavedModel exported by the EasyRec processor, and uses Blade for inference optimization on both CPUs and GPUs.

• Feature Tracking and Incremental Updates: In most cases, these modules are used for real-time training. For more information, see Online Deep Learning.

Limits

The EasyRec processor can be used on GPU devices that are of the T4, A10, 3090, and 4090 types, and general-purpose Elastic Compute Service (ECS) instance families including g6, g7, and g8 that use Intel CPUs.

Processor versions

The EasyRec processor is continuously being improved. Later versions provide enhanced features and inference performance. For optimal results, we recommend that you use the latest version to deploy your inference service. The following table lists the released versions and their basic information.

Step 1: Deploy a service

When you use the EASCMD client to deploy an EasyRec model service, you must set Processor Type to easyrec-{version}. For more information, see Deploy model services by using EASCMD or DSW. The following code provides examples on the configuration file.

Sample code when FG is enabled

bizdate=$1
cat << EOF > echo.json
{
  "name":"ali_rec_rnk_with_fg",
  "metadata": {
    "instance": 2,
    "rpc": {
      "enable_jemalloc": 1,
      "max_queue_size": 100
    }
  },
  "cloud": {
    "computing": {
      "instance_type": "ecs.g7.large"",
      "instances": null
    }
  },
  "model_config": {
    "remote_type": "hologres",
    "url": "postgresql://<AccessKeyID>:<AccessKeySecret>@<DomainName>:<port>/<database>",
    "tables": [{"name":"<schema>.<table_name>","key":"<index_column_name>","value": "<column_name>"}],
    "period": 2880,
    "fg_mode": "tf",
    "outputs":"probs_ctr,probs_cvr",
  },
  "model_path": "",
  "processor": "easyrec-1.9",
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20221122/export/final_with_fg"
      }
    }
  ]
}

EOF
# Run the deployment command. 
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# Run the update command.
eascmd update ali_rec_rnk_with_fg -s echo.json

Sample code when FG is disabled

bizdate=$1
cat << EOF > echo.json
{
  "name":"ali_rec_rnk_no_fg",
  "metadata": {
    "instance": 2,
    "rpc": {
      "enable_jemalloc": 1,
      "max_queue_size": 100
    }
  },
  "cloud": {
    "computing": {
      "instance_type": "ecs.g7.large"",
      "instances": null
    }
  },
  "model_config": {
    "fg_mode": "bypass"
  },
  "processor": "easyrec-1.9",
  "processor_envs": [
    {
      "name": "INPUT_TILE",
      "value": "2"
    }
  ],
  "storage": [
    {
      "mount_path": "/home/admin/docker_ml/workspace/model/",
      "oss": {
        "path": "oss://easyrec/ali_rec_sln_acc_rnk/20221122/export/final/"
      }
    }
  ],
  "warm_up_data_path": "oss://easyrec/ali_rec_sln_acc_rnk/rnk_warm_up.bin"
}

EOF
# Run the deployment command. 
eascmd  create echo.json
# eascmd -i <AccessKeyID>  -k  <AccessKeySecret>   -e <endpoint> create echo.json
# Run the update command.
eascmd update ali_rec_rnk_no_fg -s echo.json

The following table describes the key parameters. For information about other parameters, see Parameters of model services.

Parameters used for inference optimization of the EasyRec processor

Step 2: Call the service

After you deploy the EasyRec model service, go to the Elastic Algorithm Service (EAS) page. On this page, click Invocation Method in the Service Type column to view the endpoint and token of the service.

The input and output of the EasyRec model service are in the Protocol Buffers (protobuf) format. You can call the service in the following ways based on whether FG is enabled:

Sample code when FG is enabled (fg_mode=tf)

import com.aliyun.openservices.eas.predict.http.*;
import com.aliyun.openservices.eas.predict.request.EasyRecRequest;

PredictClient client = new PredictClient(new HttpConfig());
// Specify the endpoint of the service that you want to call. The endpoint starts with your user ID. 
client.setEndpoint("xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com");
client.setModelName("ali_rec_rnk_with_fg");
// Specify the token of the service. 
client.setToken("******");

EasyRecRequest easyrecRequest = new EasyRecRequest(separator);
// userFeatures: Specify multiple user features at the same time. Separate multiple user features with \u0002 (CTRL_B). For each user feature, separate the feature name and feature value with a colon (:). 
//  user_fea0:user_fea0_val\u0002user_fea1:user_fea1_val
// For more information about the feature value format, visit https://easyrec.readthedocs.io/en/latest/feature/rtp_fg.html.
easyrecRequest.appendUserFeatureString(userFeatures);
// Alternatively, add one user feature at a time.
// easyrecRequest.addUserFeature(String userFeaName, T userFeaValue). 
// T: the type of the feature value. Valid values: String, float, long, and int. 

// contextFeatures: Specify multiple context features at the same time. Separate multiple context features with \u0002 (CTRL_B). For each context feature, separate the feature name and feature value with a colon (:). 
//   ctxt_fea0:ctxt_fea0_ival0:ctxt_fea0_ival1:ctxt_fea0_ival2\u0002ctxt_fea1:ctxt_fea1_ival0:ctxt_fea1_ival1:ctxt_fea1_ival2
easyrecRequest.appendContextFeatureString(contextFeatures);
// Alternatively, add one context feature at a time.
// easyrecRequest.addContextFeature(String ctxtFeaName, List<Object> ctxtFeaValue). 
// Valid data types of ctxtFeaValue: String, Float, Long, and Integer. 

// itemIdStr: the list of item IDs to be predicted. Separate multiple item IDs with commas (,). 
easyrecRequest.appendItemStr(itemIdStr, ",");
// Alternatively, add one item ID at a time.
// easyrecRequest.appendItemId(String itemId)

PredictProtos.PBResponse response = client.predict(easyrecRequest);

for (Map.Entry<String, PredictProtos.Results> entry : response.getResultsMap().entrySet()) {
    String key = entry.getKey();
    PredictProtos.Results value = entry.getValue();
    System.out.print("key: " + key);
    for (int i = 0; i < value.getScoresCount(); i++) {
        System.out.format("value: %.6g\n", value.getScores(i));
    }
}

// Obtain the features processed by FG to compare with the offline features. 
// Set DebugLevel to 1 to return the generated features. 
easyrecRequest.setDebugLevel(1);
PredictProtos.PBResponse response = client.predict(easyrecRequest);
Map<String, String> genFeas = response.getGenerateFeaturesMap();
for(String itemId: genFeas.keySet()) {
    System.out.println(itemId);
    System.out.println(genFeas.get(itemId));
}

from eas_prediction import PredictClient

from eas_prediction.easyrec_request import EasyRecRequest
from eas_prediction.easyrec_predict_pb2 import PBFeature
from eas_prediction.easyrec_predict_pb2 import PBRequest

if __name__ == '__main__':
    endpoint = 'http://xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com'
    service_name = 'ali_rec_rnk_with_fg'
    token = '******'

    client = PredictClient(endpoint, service_name)
    client.set_token(token)
    client.init()

    req = PBRequest()
    uid = PBFeature()
    uid.string_feature = 'u0001'
    req.user_features['user_id'] = uid
    age = PBFeature()
    age.int_feature = 12
    req.user_features['age'] = age
    weight = PBFeature()
    weight.float_feature = 129.8
    req.user_features['weight'] = weight

    req.item_ids.extend(['item_0001', 'item_0002', 'item_0003'])
    
    easyrec_req = EasyRecRequest()
    easyrec_req.add_feed(req, debug_level=0)
    res = client.predict(easyrec_req)
    print(res)

Sample code when FG is disabled (fg_mode=bypass)

import java.util.List;

import com.aliyun.openservices.eas.predict.http.PredictClient;
import com.aliyun.openservices.eas.predict.http.HttpConfig;
import com.aliyun.openservices.eas.predict.request.TFDataType;
import com.aliyun.openservices.eas.predict.request.TFRequest;
import com.aliyun.openservices.eas.predict.response.TFResponse;

public class TestEasyRec {
    public static TFRequest buildPredictRequest() {
        TFRequest request = new TFRequest();
 
        request.addFeed("user_id", TFDataType.DT_STRING, 
                        new long[]{5}, new String []{ "u0001", "u0001", "u0001"});
      	request.addFeed("age", TFDataType.DT_FLOAT, 
                        new long[]{5}, new float []{ 18.0f, 18.0f, 18.0f});
        // Note: If you set INPUT_TILE to 2, you can simplify the code in the following way:
        //    request.addFeed("user_id", TFDataType.DT_STRING,
        //            new long[]{1}, new String []{ "u0001" });
        //    request.addFeed("age", TFDataType.DT_FLOAT, 
        //            new long[]{1}, new float []{ 18.0f});
      	request.addFeed("item_id", TFDataType.DT_STRING, 
                        new long[]{5}, new String []{ "i0001", "i0002", "i0003"});  
        request.addFetch("probs");
      	return request;
    }

    public static void main(String[] args) throws Exception {
        PredictClient client = new PredictClient(new HttpConfig());

        // Call setDirectEndpoint to access the service by using a virtual private cloud (VPC) direct connection channel. 
        //   client.setDirectEndpoint("pai-eas-vpc.cn-shanghai.aliyuncs.com");
        // You need to create a VPC direct connection channel on the EAS page of the PAI console. 
        // Compared with using a gateway, using the direct connection channel improves stability and performance. 
        client.setEndpoint("xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com");
        client.setModelName("ali_rec_rnk_no_fg");
        client.setToken("");
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 100; i++) {
            try {
                TFResponse response = client.predict(buildPredictRequest());
                // probs: the name of the output field. You can run the cURL command to view the input and output of the model.
                //   curl xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com -H "Authorization:{token}"
                List<Float> result = response.getFloatVals("probs");
                System.out.print("Predict Result: [");
                for (int j = 0; j < result.size(); j++) {
                    System.out.print(result.get(j).floatValue());
                    if (j != result.size() - 1) {
                        System.out.print(", ");
                    }
                }
                System.out.print("]\n");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        long endTime = System.currentTimeMillis();
        System.out.println("Spend Time: " + (endTime - startTime) + "ms");
        client.shutdown();
    }
}

#!/usr/bin/env python

from eas_prediction import PredictClient
from eas_prediction import StringRequest
from eas_prediction import TFRequest

if __name__ == '__main__':
    client = PredictClient('http://xxxxxxx.vpc.cn-hangzhou.pai-eas.aliyuncs.com', 'ali_rec_rnk_no_fg')
    client.set_token('')
    client.init()

    req = TFRequest('server_default') # Replace server_dafault with the signature_name of the actual model. For more information, see https://www.alibabacloud.com/help/en/pai/user-guide/sdk-for-python 
    req.add_feed('user_id', [3], TFRequest.DT_STRING, ['u0001'] * 3)
    req.add_feed('age', [3], TFRequest.DT_FLOAT, [18.0] * 3)
    # Note: If you set INPUT_TILE to 2, you can simplify the code in the following way:
    #   req.add_feed('user_id', [1], TFRequest.DT_STRING, ['u0001'])
    #   req.add_feed('age', [1], TFRequest.DT_FLOAT, [18.0])
    req.add_feed('item_id', [3], TFRequest.DT_STRING, 
        ['i0001', 'i0002', 'i0003'])
    for x in range(0, 100):
        resp = client.predict(req)
        print(resp)

You can also create custom service requests. For more information, see Request syntax.

Request syntax

For clients other than Python, you need to generate prediction code manually from the .proto file. Use the following protobuf definitions to generate code for custom service requests:

• tf_predict.proto: protobuf definition for TensorFlow models

  syntax = "proto3";
  
  option cc_enable_arenas = true;
  option go_package = ".;tf";
  option java_package = "com.aliyun.openservices.eas.predict.proto";
  option java_outer_classname = "PredictProtos";
  
  enum ArrayDataType {
    // Not a legal value for DataType. Used to indicate a DataType field
    // has not been set.
    DT_INVALID = 0;
  
    // Data types that all computation devices are expected to be
    // capable to support.
    DT_FLOAT = 1;
    DT_DOUBLE = 2;
    DT_INT32 = 3;
    DT_UINT8 = 4;
    DT_INT16 = 5;
    DT_INT8 = 6;
    DT_STRING = 7;
    DT_COMPLEX64 = 8;  // Single-precision complex
    DT_INT64 = 9;
    DT_BOOL = 10;
    DT_QINT8 = 11;     // Quantized int8
    DT_QUINT8 = 12;    // Quantized uint8
    DT_QINT32 = 13;    // Quantized int32
    DT_BFLOAT16 = 14;  // Float32 truncated to 16 bits.  Only for cast ops.
    DT_QINT16 = 15;    // Quantized int16
    DT_QUINT16 = 16;   // Quantized uint16
    DT_UINT16 = 17;
    DT_COMPLEX128 = 18;  // Double-precision complex
    DT_HALF = 19;
    DT_RESOURCE = 20;
    DT_VARIANT = 21;  // Arbitrary C++ data types
  }
  
  // Dimensions of an array
  message ArrayShape {
    repeated int64 dim = 1 [packed = true];
  }
  
  // Protocol buffer representing an array
  message ArrayProto {
    // Data Type.
    ArrayDataType dtype = 1;
  
    // Shape of the array.
    ArrayShape array_shape = 2;
  
    // DT_FLOAT.
    repeated float float_val = 3 [packed = true];
  
    // DT_DOUBLE.
    repeated double double_val = 4 [packed = true];
  
    // DT_INT32, DT_INT16, DT_INT8, DT_UINT8.
    repeated int32 int_val = 5 [packed = true];
  
    // DT_STRING.
    repeated bytes string_val = 6;
  
    // DT_INT64.
    repeated int64 int64_val = 7 [packed = true];
  
    // DT_BOOL.
    repeated bool bool_val = 8 [packed = true];
  }
  
  // PredictRequest specifies which TensorFlow model to run, as well as
  // how inputs are mapped to tensors and how outputs are filtered before
  // returning to user.
  message PredictRequest {
    // A named signature to evaluate. If unspecified, the default signature
    // will be used
    string signature_name = 1;
  
    // Input tensors.
    // Names of input tensor are alias names. The mapping from aliases to real
    // input tensor names is expected to be stored as named generic signature
    // under the key "inputs" in the model export.
    // Each alias listed in a generic signature named "inputs" should be provided
    // exactly once in order to run the prediction.
    map<string, ArrayProto> inputs = 2;
  
    // Output filter.
    // Names specified are alias names. The mapping from aliases to real output
    // tensor names is expected to be stored as named generic signature under
    // the key "outputs" in the model export.
    // Only tensors specified here will be run/fetched and returned, with the
    // exception that when none is specified, all tensors specified in the
    // named signature will be run/fetched and returned.
    repeated string output_filter = 3;
    
    // Debug flags
    // 0: just return prediction results, no debug information
    // 100: return prediction results, and save request to model_dir 
    // 101: save timeline to model_dir
    int32 debug_level = 100;
  }
  
  // Response for PredictRequest on successful run.
  message PredictResponse {
    // Output tensors.
    map<string, ArrayProto> outputs = 1;
  }

• easyrec_predict.proto: protobuf definition for TensorFlow models and FG

  syntax = "proto3";
  
  option cc_enable_arenas = true;
  option go_package = ".;easyrec";
  option java_package = "com.aliyun.openservices.eas.predict.proto";
  option java_outer_classname = "EasyRecPredictProtos";
  
  import "tf_predict.proto";
  
  // context features
  message ContextFeatures {
    repeated PBFeature features = 1;
  }
  
  message PBFeature {
    oneof value {
      int32 int_feature = 1;
      int64 long_feature = 2;
      string string_feature = 3;
      float float_feature = 4;
    }
  }
  
  // PBRequest specifies the request for aggregator
  message PBRequest {
    // Debug flags
    // 0: just return prediction results, no debug information
    // 3: return features generated by FG module, string format, feature values are separated by \u0002, 
    //    could be used for checking feature consistency check and generating online deep learning samples 
    // 100: return prediction results, and save request to model_dir 
    // 101: save timeline to model_dir
    // 102: for recall models such as DSSM and MIND, only only return Faiss retrieved results
    //      but also return user embedding vectors.
    int32 debug_level = 1;
  
    // user features
    map<string, PBFeature> user_features = 2;
  
    // item ids, static(daily updated) item features 
    // are fetched from the feature cache resides in 
    // each processor node by item_ids
    repeated string item_ids = 3;
  
    // context features for each item, realtime item features
    //    could be passed as context features.
    map<string, ContextFeatures> context_features = 4;
  
    // embedding retrieval neighbor number.
    int32 faiss_neigh_num = 5;
  }
  
  // return results
  message Results {
    repeated double scores = 1 [packed = true];
  }
  
  enum StatusCode {
    OK = 0;
    INPUT_EMPTY = 1;
    EXCEPTION = 2;
  }
  
  // PBResponse specifies the response for aggregator
  message PBResponse {
    // results
    map<string, Results> results = 1;
  
    // item features
    map<string, string> item_features = 2;
  
    // fg generate features
    map<string, string> generate_features = 3;
  
    // context features
    map<string, ContextFeatures> context_features = 4;
  
    string error_msg = 5;
  
    StatusCode status_code = 6;
  
    // item ids
    repeated string item_ids = 7;
  
    repeated string outputs = 8;
  
    // all fg input features
    map<string, string> raw_features = 9;
  
    // output tensors
    map<string, ArrayProto> tf_outputs = 10;
  }