Mit Iteratoren

Milvus bietet Such- und Abfrage-Iteratoren für die Iteration von Ergebnissen mit einer großen Anzahl von Entitäten. Da Milvus TopK auf 16384 begrenzt, können Benutzer Iteratoren verwenden, um große Zahlen oder sogar ganze Entitäten in einer Sammlung im Batch-Modus zurückzugeben.

Überblick

Iteratoren sind leistungsstarke Werkzeuge, mit denen Sie eine große Menge von Daten oder alle Daten innerhalb einer Sammlung mit Hilfe von Primärschlüsselwerten und booleschen Ausdrücken durchlaufen können. Dies kann die Art und Weise, wie Sie Daten abrufen, erheblich verbessern. Im Gegensatz zur herkömmlichen Verwendung von Offset- und Limit-Parametern, die mit der Zeit an Effizienz verlieren können, bieten Iteratoren eine besser skalierbare Lösung.

Vorteile der Verwendung von Iteratoren

• Vereinfachung: Komplexe Offset- und Grenzwerteinstellungen entfallen.

• Effizient: Skalierbarer Datenabruf, da nur die benötigten Daten abgerufen werden.

• Konsistenz: Gewährleistet eine konsistente Datensatzgröße mit booleschen Filtern.

Vorbereitungen

In den folgenden Schritten wird der Code für die Verbindung mit Milvus, die schnelle Einrichtung einer Sammlung und das Einfügen von über 10.000 zufällig generierten Entitäten in die Sammlung wiederholt.

Schritt 1: Erstellen einer Sammlung

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);

Schritt 2: Zufällig generierte Entitäten einfügen

# 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());

Suche mit Iterator

Iteratoren machen Ähnlichkeitssuchen besser skalierbar.

1. Initialisieren Sie den Such-Iterator, um die Suchparameter und Ausgabefelder zu definieren.

2. Verwenden Sie die next() -Methode innerhalb einer Schleife, um durch die Suchergebnisse zu paginieren.

   • Wenn die Methode ein leeres Array zurückgibt, endet die Schleife, und es sind keine weiteren Seiten verfügbar.

   • Alle Ergebnisse enthalten die angegebenen Ausgabefelder.

3. Rufen Sie manuell die Methode close() auf, um den Iterator zu schließen, sobald alle Daten abgerufen wurden.

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());

Parameter	Beschreibung
data	Eine Liste von Vektoreinbettungen. Milvus sucht nach den Vektoreinbettungen, die den angegebenen am ähnlichsten sind.
anns_field	Der Name des Vektorfeldes in der aktuellen Sammlung.
batch_size	Die Anzahl der Entitäten, die jedes Mal zurückgegeben werden sollen, wenn Sie next() für den aktuellen Iterator aufrufen. Der Wert ist standardmäßig 1000. Setzen Sie ihn auf einen geeigneten Wert, um die Anzahl der pro Iteration zurückzugebenden Objekte zu steuern.
param	Die spezifischen Parametereinstellungen für diesen Vorgang. metric_type: Der Metrik-Typ, der auf diese Operation angewendet wird. Dies sollte derselbe sein, der verwendet wird, wenn Sie das oben angegebene Vektorfeld indizieren. Mögliche Werte sind L2, IP, COSINE, JACCARD, HAMMING. params: Zusätzliche Parameter. Einzelheiten finden Sie unter search_iterator().
output_fields	Eine Liste von Feldnamen, die in jeder zurückgegebenen Entität enthalten sein sollen. Der Standardwert ist None. Wenn er nicht angegeben wird, wird nur das Primärfeld einbezogen.
limit	Die Gesamtzahl der zurückzugebenden Entitäten. Der Standardwert ist -1, was bedeutet, dass alle passenden Entitäten zurückgegeben werden.

Parameter	Beschreibung
withCollectionName	Legt den Namen der Sammlung fest. Der Sammlungsname darf nicht leer oder null sein.
withVectorFieldName	Zielvektorfeld nach Name festlegen. Der Feldname darf nicht leer oder ungültig sein.
withVectors	Legen Sie die Zielvektoren fest. Bis zu 16384 Vektoren sind zulässig.
withBatchSize	Die Anzahl der Entitäten, die bei jedem Aufruf von next() für den aktuellen Iterator zurückgegeben werden. Der Standardwert ist 1000. Setzen Sie ihn auf einen geeigneten Wert, um die Anzahl der Entitäten zu steuern, die pro Iteration zurückgegeben werden.
withParams	Gibt die Parameter der Suche im JSON-Format an. Weitere Informationen finden Sie unter searchIterator().

Abfrage mit einem Iterator

# 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();
    }
}

Parameter	Beschreibung
batch_size	Die Anzahl der Entitäten, die jedes Mal zurückgegeben werden sollen, wenn Sie next() für den aktuellen Iterator aufrufen. Der Standardwert ist 1000. Setzen Sie ihn auf einen geeigneten Wert, um die Anzahl der Entitäten zu steuern, die pro Iteration zurückgegeben werden sollen.
expr	Eine skalare Filterbedingung, um übereinstimmende Entitäten zu filtern. Der Standardwert ist None, was anzeigt, dass die skalare Filterung ignoriert wird. Um eine skalare Filterbedingung zu erstellen, siehe Boolesche Ausdrucksregeln.
output_fields	Eine Liste von Feldnamen, die in jeder zurückgegebenen Entität enthalten sein sollen. Der Standardwert ist None. Wenn er nicht angegeben wird, wird nur das Primärfeld einbezogen.
limit	Die Gesamtzahl der zurückzugebenden Entitäten. Der Standardwert ist -1, was bedeutet, dass alle passenden Entitäten zurückgegeben werden.

Parameter	Beschreibung
withCollectionName	Legt den Namen der Sammlung fest. Der Sammlungsname darf nicht leer oder null sein.
withExpr	Legen Sie den Ausdruck für die Abfrage von Entitäten fest. Um eine skalare Filterbedingung zu erstellen, siehe Boolesche Ausdrucksregeln.
withBatchSize	Die Anzahl der Entitäten, die bei jedem Aufruf von next() für den aktuellen Iterator zurückgegeben werden sollen. Der Standardwert ist 1000. Setzen Sie ihn auf einen geeigneten Wert, um die Anzahl der pro Iteration zurückzugebenden Entitäten zu steuern.
addOutField	Gibt ein skalares Ausgabefeld an (Optional).