View a markdown version of this page

Streaming-Abfrageergebnisse mit Gremlin - Amazon Neptune
Schrittweise Nutzung der ErgebnisseEifriger Konsum versus 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.

Streaming-Abfrageergebnisse mit Gremlin

Wenn Sie eine Gremlin-Traversal ausführen, die 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 mehr anfordert. Dies kann von Vorteil sein, wenn Sie Ergebnisse so verarbeiten möchten, wie sie vom Server zurückgegeben werden, erfordert jedoch die Verwendung von verzögerten Iterationsmustern, um zu vermeiden, dass die gesamte Ergebnismenge im Speicher gesammelt wird.

Neptune sendet standardmäßig Ergebnisse in Batches von 64 pro WebSocket Frame. Sie können diese serverseitige Standardeinstellung nicht ändern, aber die Batchgröße kann pro Anfrage vom Client mithilfe der Request-Option (aufgerufen Tokens.ARGS_BATCH_SIZE im Java-Treiber oder batchSizeals Standard auf Treiberebene) außer Kraft gesetzt werden. connectionPool.resultIterationBatchSize

Einzelheiten zur Konfiguration von Treibern batchSize in anderen Sprachen 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 weiterleitet, wird der clientseitige Gegendruck implizit über TCP und Flusssteuerung abgewickelt. WebSocket Wenn der Client 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. Durchläufe, dieorder(),,groupCount(), oder andere Schritte beinhalten group()dedup(), bei denen die vollständige Durchquerung abgeschlossen sein muss, bevor Ergebnisse ausgegeben werden, veranlassen Neptune, die gesamte Ergebnismenge im Speicher zu materialisieren, bevor das Streaming beginnt. In diesen Fällen reduziert die Batchverarbeitung immer noch den Aufwand für die Serialisierung pro Frame, reduziert jedoch nicht die serverseitige Speichernutzung.

Schrittweise Nutzung der Ergebnisse

Um Ergebnisse sofort zu verarbeiten, iterieren Sie langsam, indem SiehasNext()/next()oder etwas Ähnliches verwenden, APIs anstatt alle Ergebnisse in einer Liste zu sammeln. Sie können next(batchSize) es verwenden, um Ergebnisse in Batches auf Anwendungsebene abzurufen, sodass Sie Zwischenarbeiten zwischen den Batches 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 Konsum versus inkrementeller Konsum

Streaming ermöglicht es Ihnen, Ergebnisse inkrementell zu verarbeiten, wenn 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 die Ergebnisse reagiert, sobald sie eintreffen:

  • Java: toList() oder toSet()

  • Python: toList() oder toSet()

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

  • .NET: ToList() oder Promise()

  • Node.js: toList()

Anmerkung

Selbst bei Verwendung dieser Methoden werden Daten immer noch inkrementell über die WebSocket Verbindung übertragen. 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.