View a markdown version of this page

Streamen von Abfrageergebnissen mit Gremlin - Amazon Neptune
Inkrementelle Nutzung der ErgebnisseEifriger oder inkrementeller Konsum

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Streamen von Abfrageergebnissen mit Gremlin

Wenn Sie einen Gremlin-Traversal ausführen, der eine große Anzahl von Ergebnissen zurückgibt, streamt Neptune diese stapelweise über die Verbindung zurück an den Client. WebSocket Neptune sendet Ergebnisstapel, sobald sie produziert werden, ohne darauf zu warten, dass der Kunde weitere anfordert. Dies kann von Vorteil sein, wenn Sie Ergebnisse so verarbeiten möchten, wie sie vom Server zurückgegeben werden. Sie müssen jedoch verzögerte Iterationsmuster verwenden, um zu vermeiden, dass der gesamte Ergebnissatz im Speicher gespeichert wird.

Neptune sendet die Ergebnisse standardmäßig in Batches von 64 pro Frame. WebSocket Sie können diesen serverseitigen Standard nicht ändern, aber die Batchgröße kann vom Client aus mithilfe der Anforderungsoption (die Tokens.ARGS_BATCH_SIZE im Java-Treiber oder als Standardeinstellung auf Treiberebene aufgerufen wird) für jede batchSize Anforderung außer Kraft gesetzt werden. connectionPool.resultIterationBatchSize

Einzelheiten zur Konfiguration batchSize in anderen Sprachtreibern finden Sie im Abschnitt Konfiguration für jeden Treiber in der Dokumentation zu Apache TinkerPop Gremlin Drivers and Variants.

Da der Server die Ergebnisse automatisch überträgt, wird der clientseitige Gegendruck implizit über TCP und Flusskontrolle abgewickelt. WebSocket Wenn der Client nur langsam aus dem Socket liest, werden die Schreibvorgänge des Servers irgendwann blockiert, bis der Client aufholt.

Wichtig

Streaming ist am effektivsten bei Durchläufen, bei denen schrittweise Ergebnisse erzielt werden können. Bei Durchläufen, dieorder(),groupCount(), oder andere Schritte beinhalten group()dedup(), bei denen der gesamte Durchlauf abgeschlossen sein muss, bevor Ergebnisse ausgegeben werden, materialisiert Neptune den gesamten Ergebnissatz im Speicher, bevor das Streaming beginnt. In diesen Fällen reduziert die Stapelverarbeitung immer noch den Serialisierungsaufwand pro Frame, reduziert jedoch nicht die serverseitige Speicherauslastung.

Inkrementelle Nutzung der Ergebnisse

Um die Ergebnisse sofort zu verarbeiten, iterieren Sie träge mithilfe vonhasNext()/next()oder gleichwertigen APIs, anstatt alle Ergebnisse in einer Liste zusammenzufassen. Sie können next(batchSize) es verwenden, um Ergebnisse in Batches auf Anwendungsebene abzurufen, sodass Sie zwischen den Batches Zwischenarbeiten durchführen können, während der Server weiterhin Ergebnisse produziert.

Beispiel Java (GLV-Bytecode)
GraphTraversalSource g = traversal().withRemote(connection);

int batchSize = 10;
int batchNum = 0;
var traversal = g.V().hasLabel("movie").values("title").limit(1000);
while (traversal.hasNext()) {
    var batch = traversal.next(batchSize);
    batchNum++;
    for (var title : batch) {
        System.out.println("  " + title);
    }

    // Do other intermediary work here between batch calls
    System.out.println("Batch " + batchNum + " processing complete\n");
}
Beispiel Python
g = traversal().with_remote(connection)

BATCH_SIZE = 10
batch_num = 0
t = g.V().has_label('movie').values('title').limit(1000)
while t.has_next():
    batch = t.next(BATCH_SIZE)
    batch_num += 1
    for title in batch:
        print(f"  {title}")

    # Do other intermediary work here between batch calls
    print(f"Batch {batch_num} processing complete\n")
Beispiel Go
// The Go driver does not support next(n), so batches are accumulated manually.
g := gremlingo.Traversal_().WithRemote(connection)

resultSet, err := g.V().HasLabel("movie").Values("title").Limit(1000).GetResultSet()
if err != nil {
    log.Fatal(err)
}

batchSize := 10
batchNum := 0
for {
    var batch []interface{}
    for i := 0; i < batchSize; i++ {
        result, ok, err := resultSet.One() // returns (value, ok, error); ok is false when results are exhausted
        if err != nil {
            log.Fatal(err)
        }
        if !ok {
            break
        }
        batch = append(batch, result)
    }
    if len(batch) == 0 {
        break
    }
    batchNum++
    for _, v := range batch {
        fmt.Printf("  %v\n", v)
    }

    // Do other intermediary work here between batch calls
    fmt.Printf("Batch %d processing complete\n\n", batchNum)
}
Beispiel.NET
var g = Traversal().WithRemote(connection);

var batchSize = 10;
var batchNum = 0;
var traversal = g.V().HasLabel("movie").Values<string>("title").Limit<string>(1000);
while (traversal.HasNext())
{
    var batch = traversal.Next(batchSize);
    batchNum++;
    foreach (var title in batch)
    {
        Console.WriteLine($"  {title}");
    }

    // Do other intermediary work here between batch calls
    Console.WriteLine($"Batch {batchNum} processing complete\n");
}
Beispiel Node.js
// The Node.js driver does not support next(n), so batches are accumulated manually.
const g = traversal().withRemote(connection);

const batchSize = 10;
let batchNum = 0;
const t = g.V().hasLabel('movie').values('title').limit(1000);
while (true) {
    const batch = [];
    for (let i = 0; i < batchSize; i++) {
        const result = await t.next();
        if (result.done) break;
        batch.push(result.value);
    }
    if (batch.length === 0) break;
    batchNum++;
    for (const title of batch) {
        console.log(`  ${title}`);
    }

    // Do other intermediary work here between batch calls
    console.log(`Batch ${batchNum} processing complete\n`);
}

Eifriger oder inkrementeller Konsum

Streaming ermöglicht es Ihnen, Ergebnisse schrittweise zu verarbeiten, während zusätzliche Daten abgerufen und zurückgegeben werden. Die folgenden Methoden blockieren, bis die gesamte Ergebnismenge im Speicher erfasst ist. Dadurch wird verhindert, dass Ihre Anwendung auf Ergebnisse reagiert, sobald sie eintreffen:

  • Java: toList() oder toSet()

  • Python: toList() oder toSet()

  • Gehe: ToList()ToSet(), oder GetResultSet().GetAll()

  • .NET: ToList() oder Promise()

  • Node.js: toList()

Anmerkung

Daten werden auch dann schrittweise über die WebSocket Verbindung übertragen, wenn diese Methoden verwendet werden. Der Unterschied besteht darin, dass Ihre Anwendung einzelne Ergebnisse erst verarbeiten kann, wenn die gesamte Erfassung abgeschlossen ist. Verwenden Sie die in den obigen Beispielen gezeigten Lazy-Iterations- oder Batch-Muster, um die Ergebnisse sofort zu verarbeiten.