使用迭代器
Milvus 提供搜索和查询迭代器，用于迭代大量实体的结果。由于 Milvus 将 TopK 限制在 16384，用户可以使用迭代器在批处理模式下返回大量甚至整个集合中的实体。
概述
迭代器是一种功能强大的工具，能帮助你使用主键值和布尔表达式迭代大量数据或集合中的所有数据。这可以大大改进检索数据的方式。传统的偏移和限制参数会随着时间的推移而降低效率，而迭代器则不同，它提供了一种更具可扩展性的解决方案。
使用迭代器的好处
简单：消除了复杂的偏移和限制设置。
高效：只获取需要的数据，提供可扩展的数据检索。
一致性通过布尔筛选器确保数据集大小一致。
注释
- 此功能适用于 Milvus 2.3.x 或更高版本。
准备工作
以下步骤重新利用代码连接到 Milvus，快速建立一个数据集，并在数据集中插入超过 10,000 个随机生成的实体。
步骤 1：创建集合
使用 MilvusClient连接到 Milvus 服务器，并使用 create_collection()来创建集合。
使用 MilvusClientV2连接到 Milvus 服务器并 createCollection()创建集合。
from pymilvus import MilvusClient
# 1. Set up a Milvus client
client = MilvusClient(
uri="http://localhost:19530"
)
# 2. Create a collection
client.create_collection(
collection_name="quick_setup",
dimension=5,
)
import io.milvus.client.MilvusServiceClient;
import io.milvus.param.ConnectParam;
import io.milvus.param.highlevel.collection.CreateSimpleCollectionParam;
String CLUSTER_ENDPOINT = "http://localhost:19530";
// 1. Connect to Milvus server
ConnectParam connectParam = ConnectParam.newBuilder()
.withUri(CLUSTER_ENDPOINT)
.build();
MilvusServiceClient client = new MilvusServiceClient(connectParam);
// 2. Create a collection
CreateSimpleCollectionParam createCollectionParam = CreateSimpleCollectionParam.newBuilder()
.withCollectionName("quick_setup")
.withDimension(5)
.build();
client.createCollection(createCollectionParam);
第二步：插入随机生成的实体
使用 insert()将实体插入集合。
使用 insert()将实体插入集合。
# 3. Insert randomly generated vectors
colors = ["green", "blue", "yellow", "red", "black", "white", "purple", "pink", "orange", "brown", "grey"]
data = []
for i in range(10000):
current_color = random.choice(colors)
current_tag = random.randint(1000, 9999)
data.append({
"id": i,
"vector": [ random.uniform(-1, 1) for _ in range(5) ],
"color": current_color,
"tag": current_tag,
"color_tag": f"{current_color}_{str(current_tag)}"
})
print(data[0])
# Output
#
# {
# "id": 0,
# "vector": [
# -0.5705990742218152,
# 0.39844925120642083,
# -0.8791287928610869,
# 0.024163154953680932,
# 0.6837669917169638
# ],
# "color": "purple",
# "tag": 7774,
# "color_tag": "purple_7774"
# }
res = client.insert(
collection_name="quick_setup",
data=data,
)
print(res)
# Output
#
# {
# "insert_count": 10000,
# "ids": [
# 0,
# 1,
# 2,
# 3,
# 4,
# 5,
# 6,
# 7,
# 8,
# 9,
# "(9990 more items hidden)"
# ]
# }
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import com.alibaba.fastjson.JSONObject;
import io.milvus.param.R;
import io.milvus.param.dml.InsertParam;
import io.milvus.response.MutationResultWrapper;
import io.milvus.grpc.MutationResult;
// 3. Insert randomly generated vectors into the collection
List<String> colors = Arrays.asList("green", "blue", "yellow", "red", "black", "white", "purple", "pink", "orange", "brown", "grey");
List<JSONObject> data = new ArrayList<>();
for (int i=0; i<10000; i++) {
Random rand = new Random();
String current_color = colors.get(rand.nextInt(colors.size()-1));
JSONObject row = new JSONObject();
row.put("id", Long.valueOf(i));
row.put("vector", Arrays.asList(rand.nextFloat(), rand.nextFloat(), rand.nextFloat(), rand.nextFloat(), rand.nextFloat()));
row.put("color_tag", current_color + "_" + String.valueOf(rand.nextInt(8999) + 1000));
data.add(row);
}
InsertParam insertParam = InsertParam.newBuilder()
.withCollectionName("quick_setup")
.withRows(data)
.build();
R<MutationResult> insertRes = client.insert(insertParam);
if (insertRes.getStatus() != R.Status.Success.getCode()) {
System.err.println(insertRes.getMessage());
}
MutationResultWrapper wrapper = new MutationResultWrapper(insertRes.getData());
System.out.println(wrapper.getInsertCount());
使用迭代器搜索
迭代器使相似性搜索更具可扩展性。
要使用迭代器搜索，请调用search_iterator()方法：
要使用迭代器搜索，请调用searchIterator()方法：
初始化搜索迭代器，定义搜索参数和输出字段。
在循环中使用next()方法对搜索结果进行分页。
如果该方法返回一个空数组，则循环结束，不再提供更多页面。
所有结果都包含指定的输出字段。
检索完所有数据后，手动调用close()方法关闭迭代器。
from pymilvus import Collection
# 4. Search with iterator
connections.connect(host="127.0.0.1", port=19530)
collection = Collection("quick_setup")
query_vectors = [[0.3580376395471989, -0.6023495712049978, 0.18414012509913835, -0.26286205330961354, 0.9029438446296592]]
search_params = {
"metric_type": "IP",
"params": {"nprobe": 10}
}
iterator = collection.search_iterator(
data=query_vectors,
anns_field="vector",
batch_size=10,
param=search_params,
output_fields=["color_tag"],
limit=3
)
results = []
while True:
result = iterator.next()
if not result:
iterator.close()
break
results.extend(result)
for hit in result:
results.append(hit.to_dict())
print(results)
# Output
#
# [
# {
# "id": 1756,
# "distance": 2.0642056465148926,
# "entity": {
# "color_tag": "black_9109"
# }
# },
# {
# "id": 6488,
# "distance": 1.9437453746795654,
# "entity": {
# "color_tag": "purple_8164"
# }
# },
# {
# "id": 3338,
# "distance": 1.9107104539871216,
# "entity": {
# "color_tag": "brown_8121"
# }
# }
# ]
import io.milvus.param.dml.QueryIteratorParam;
import io.milvus.param.dml.SearchIteratorParam;
import io.milvus.response.QueryResultsWrapper;
import io.milvus.orm.iterator.SearchIterator;
// 4. Search with iterators
SearchIteratorParam iteratorParam = SearchIteratorParam.newBuilder()
.withCollectionName("quick_setup")
.withVectorFieldName("vector")
// Use withFloatVectors() in clusters compatible with Milvus 2.4.x
.withVectors(Arrays.asList(0.3580376395471989f, -0.6023495712049978f, 0.18414012509913835f, -0.26286205330961354f, 0.9029438446296592f))
.withBatchSize(10L)
.withParams("{\"metric_type\": \"COSINE\", \"params\": {\"level\": 1}}")
.build();
R<SearchIterator> searchIteratorRes = client.searchIterator(iteratorParam);
if (searchIteratorRes.getStatus() != R.Status.Success.getCode()) {
System.err.println(searchIteratorRes.getMessage());
}
SearchIterator searchIterator = searchIteratorRes.getData();
List<QueryResultsWrapper.RowRecord> results = new ArrayList<>();
while (true) {
List<QueryResultsWrapper.RowRecord> batchResults = searchIterator.next();
if (batchResults.isEmpty()) {
searchIterator.close();
break;
}
for (QueryResultsWrapper.RowRecord rowRecord : batchResults) {
results.add(rowRecord);
}
}
System.out.println(results.size());
| 参数 | 描述 |
|---|---|
data |
向量嵌入的列表。 Milvus 会搜索与指定向量嵌入最相似的向量嵌入。 |
anns_field |
当前集合中的向量字段名称。 |
batch_size |
每次在当前迭代器上调用next() 时要返回的实体数量。默认值为1000。将其设置为适当的值，以控制每次迭代返回的实体数量。 |
param |
此操作的特定参数设置。
|
output_fields |
要包含在返回的每个实体中的字段名列表。 默认值为"无"。如果未指定，则只包含主字段。 |
limit |
要返回的实体总数。 默认值为-1，表示将返回所有匹配实体。 |
| 参数 | 说明 |
|---|---|
withCollectionName |
设置集合名称。集合名称不能为空或空值。 |
withVectorFieldName |
按名称设置目标向量字段。字段名称不能为空或空。 |
withVectors |
设置目标向量。最多允许 16384 个向量。 |
withBatchSize |
每次在当前迭代器上调用next() 时返回的实体数量。默认值为1000。将其设置为适当的值，以控制每次迭代返回的实体数量。 |
withParams |
以 JSON 格式指定搜索参数。更多信息，请参阅searchIterator()。 |
使用迭代器查询
要使用迭代器查询，请调用query_iterator()方法：
要使用迭代器搜索，请调用queryIterator()方法：
# 6. Query with iterator
iterator = collection.query_iterator(
batch_size=10, # Controls the size of the return each time you call next()
expr="color_tag like \"brown_8\"",
output_fields=["color_tag"]
)
results = []
while True:
result = iterator.next()
if not result:
iterator.close()
break
results.extend(result)
# 8. Check the search results
print(len(results))
print(results[:3])
# Output
#
# [
# {
# "color_tag": "brown_8785",
# "id": 94
# },
# {
# "color_tag": "brown_8568",
# "id": 176
# },
# {
# "color_tag": "brown_8721",
# "id": 289
# }
# ]
import io.milvus.param.dml.QueryIteratorParam;
import io.milvus.orm.iterator.QueryIterator;
// 5. Query with iterators
try {
Files.write(Path.of("results.json"), JSON.toJSONString(new ArrayList<>()).getBytes(), StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
} catch (Exception e) {
// TODO: handle exception
e.printStackTrace();
}
QueryIteratorParam queryIteratorParam = QueryIteratorParam.newBuilder()
.withCollectionName("quick_setup")
.withExpr("color_tag like \"brown_8%\"")
.withBatchSize(50L)
.addOutField("vector")
.addOutField("color_tag")
.build();
R<QueryIterator> queryIteratRes = client.queryIterator(queryIteratorParam);
if (queryIteratRes.getStatus() != R.Status.Success.getCode()) {
System.err.println(queryIteratRes.getMessage());
}
QueryIterator queryIterator = queryIteratRes.getData();
while (true) {
List<QueryResultsWrapper.RowRecord> batchResults = queryIterator.next();
if (batchResults.isEmpty()) {
queryIterator.close();
break;
}
String jsonString = "";
List<JSONObject> jsonObject = new ArrayList<>();
try {
jsonString = Files.readString(Path.of("results.json"));
jsonObject = JSON.parseArray(jsonString).toJavaList(null);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
for (QueryResultsWrapper.RowRecord queryResult : batchResults) {
JSONObject row = new JSONObject();
row.put("id", queryResult.get("id"));
row.put("vector", queryResult.get("vector"));
row.put("color_tag", queryResult.get("color_tag"));
jsonObject.add(row);
}
try {
Files.write(Path.of("results.json"), JSON.toJSONString(jsonObject).getBytes(), StandardOpenOption.WRITE);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
| 参数 | 说明 |
|---|---|
batch_size |
每次在当前迭代器上调用next() 时要返回的实体数量。该值默认为1000。将其设置为合适的值，以控制每次迭代返回的实体数。 |
expr |
用于过滤匹配实体的标量过滤条件。 该值默认为 "无"，表示忽略标量过滤。要创建标量过滤条件，请参阅布尔表达式规则。 |
output_fields |
要包含在返回的每个实体中的字段名列表。 该值默认为"无"。如果未指定，则只包含主字段。 |
limit |
要返回的实体总数。 默认值为-1，表示将返回所有匹配实体。 |
| 参数 | 说明 |
|---|---|
withCollectionName |
设置集合名称。集合名称不能为空或空值。 |
withExpr |
设置查询实体的表达式。要建立标量过滤条件，请参阅布尔表达式规则。 |
withBatchSize |
每次在当前迭代器上调用next() 时要返回的实体数量。默认值为1000。将其设置为适当的值，以控制每次迭代返回的实体数。 |
addOutField |
指定输出标量字段（可选）。 |
