Java SDK 参考

接口

RecallEngineClient client = new RecallEngineClient(endpoint, username, password);

参数说明

召回引擎服务的访问地址可以在PAI-Rec管控台 召回管理基本信息页面查看。

在该页面的VPC网络高速连通配置区域，服务域名列即为召回引擎服务的访问地址（endpoint）。

内网调用-可选配置方法

说明

• 客户端默认连接超时为200ms，读写超时为500ms，连接池最大连接数为1000。

• 所有可选配置方法支持链式调用。

import com.aliyun.openservices.pairec.recallengine.RecallEngineClient;
// 通过环境变量获取认证信息
String endpoint = System.getenv("RECALL_ENGINE_SERVICE_ENDPOINT");
String username = System.getenv("RECALL_ENGINE_SERVICE_USERNAME");
String password = System.getenv("RECALL_ENGINE_SERVICE_PASSWORD");
// 创建客户端
RecallEngineClient client = new RecallEngineClient(endpoint, username, password);
// 设置重试次数
client.withRetryTimes(2);

公网调用-可选配置方法

说明

• 需要先设置ak

• 当使用公网地址时，client的endpoint可以传空值

• 将withPublicEndpoint(boolean usePublicEndpoint) 放在最后一个设置

import com.aliyun.openservices.pairec.recallengine.RecallEngineClient;
// 通过环境变量获取认证信息
String akId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
String akSecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
String region = System.getenv("PAIRC_REGION");
String username = System.getenv("RECALL_ENGINE_SERVICE_USERNAME");
String password = System.getenv("RECALL_ENGINE_SERVICE_PASSWORD");
// 创建客户端
RecallEngineClient client = new RecallEngineClient("", username, password);
// 设置公网调用参数
client.withAccessKeyId(akId)
  .withAccessKeySecret(akSecret)
  .withRegion(region)
  .withInstanceId(instanceId)
  .withPublicEndpoint(true)
  .withRetryTimes(2);

接口

public RecallResponse recall(RecallRequest request) throws RecallEngineException;

RecallRequest参数说明

RecallConf参数说明

RecallResponse说明

示例

import com.aliyun.openservices.pairec.recallengine.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
RecallEngineClient client = new RecallEngineClient(endpoint, username, password);
// 构造召回请求
RecallRequest request = new RecallRequest();
request.setInstanceId("your-instance-id");
request.setService("recall_test");
request.setVersion("V1");
request.setUid("123");
request.setRetainFields(Arrays.asList("click"))
// 设置召回配置
Map<String, RecallConf> recalls = new HashMap<>();
recalls.put("u2i_recall", new RecallConf("123", 100));
request.setRecalls(recalls);
// 执行召回
RecallResponse resp = client.recall(request);
Record result = resp.getResult();
// 获取结果信息
System.out.println("Total records: " + result.size());
System.out.println("Field names: " + result.fieldNames());

接口

public WriteResponse write(String instanceId, String table, WriteRequest request) throws RecallEngineException;

参数说明

WriteRequest参数说明

WriteResponse说明

WriteResponse 继承自 Response，包含以下字段：

示例

import com.aliyun.openservices.pairec.recallengine.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
//1. 插入包含所有字段的初始记录
WriteRequest insertRequest = new WriteRequest();
insertRequest.setRequestId("partial-update-init");
insertRequest.setInsertMode(InsertMode.INSERT);
Map<String, Object> initialData = new HashMap<>();
initialData.put("item_id", "item_partial_test_002");
initialData.put("bool_field", true);
initialData.put("category", "electronics");
initialData.put("list", java.util.Arrays.asList("tag1", "tag2", "tag3"));
java.util.List<Map<String, Object>> content = new java.util.ArrayList<>();
content.add(initialData);
insertRequest.setContent(content);
WriteResponse initResp = client.write(instanceId, tableName, insertRequest);
assertEquals("OK", initResp.getCode());
// 刷新以确保数据已写入
client.writeFlush();
System.out.println("Initial insert completed");
//2. 部分更新 — 仅更新 category（分类）字段
WriteRequest updateRequest = new WriteRequest();
updateRequest.setRequestId("partial-update-category");
updateRequest.setInsertMode(InsertMode.UPSERT);  // 支持部分字段更新的 UPSERT 操作
Map<String, Object> partialData = new HashMap<>();
// 必须包含主键以标识记录
partialData.put("item_id", "item_partial_test_002");
// 仅更新 category 字段，其他字段（bool_field、list）保持不变
partialData.put("category", "home_appliances");
java.util.List<Map<String, Object>> updateContent = new java.util.ArrayList<>();
updateContent.add(partialData);
updateRequest.setContent(updateContent);
WriteResponse updateResp = client.write(instanceId, tableName, updateRequest);
assertEquals("OK", updateResp.getCode());
// 刷新以确保数据已写入
client.writeFlush();
System.out.println("Partial update completed: category updated to 'home_appliances'");

接口

public  DeleteResponse delete(String instanceId, String table, DeleteRequest request) throws RecallEngineException

参数说明

DeleteRequest参数说明

DeleteResponse

DeleteResponse 继承自 Response，包含以下字段：

示例

import com.aliyun.openservices.pairec.recallengine.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
String instanceId = "learn-pairec-cn-3ic4mhhvr01";
String tableName = "test_exposure1";
DeleteRequest deleteRequest = new DeleteRequest();
deleteRequest.setRequestId("delete-version-test-001");
deleteRequest.setKeys(Arrays.asList("12345", "67890"));
deleteRequest.setSchema("default");
deleteRequest.setVersionId("20260327102603");
DeleteResponse deleteResp = client.delete(instanceId, tableName, deleteRequest);
System.out.println("Delete with version response: " + deleteResp.getCode() + ", " + deleteResp.getMessage());
    

接口

public Record sort(String name, boolean desc);
public Record retain(int count);
public Record filter(String name);
public Record filterByColumnValue(String name, Predicate<Object> predicate);
public Record filterByValues(Predicate<Map<String, Object>> predicate);
public Record random();
public List<Object> columnValues(String columnName);
public List<String> columnValuesString(String columnName);
public List<String> fieldNames();
public int size();

方法说明

说明

• sort、retain、filter、filterByColumnValue、filterByValues、random 方法均返回 this，支持链式调用。

• Record 不是线程安全的，请勿在多线程环境下并发操作同一个 Record 实例。

示例

RecallResponse resp = client.recall(request);
Record result = resp.getResult();
// 按 score 列降序排序，去重后取前10条
Record processed = result
        .sort("score", true)
        .filter("item_id")
        .retain(10);
System.out.println("Processed records: " + processed.size());
// 获取 item_id 列的所有值
List<String> itemIds = processed.columnValuesString("item_id");
System.out.println("Item IDs: " + itemIds);
// 按条件过滤：仅保留 score > 0.5 的记录
Record filtered = result.filterByColumnValue("score", value -> {
    if (value instanceof Number) {
        return ((Number) value).doubleValue() > 0.5;
    }
    return false;
});
// 按全行条件过滤
Record filtered2 = result.filterByValues(row -> {
    Object category = row.get("category");
    return category != null && "video".equals(category.toString());
});