View a markdown version of this page

Diffusion en continu des résultats de requêtes avec Gremlin - Amazon Neptune
Consommer les résultats de manière incrémentielleConsommation rapide ou consommation incrémentielle

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Diffusion en continu des résultats de requêtes avec Gremlin

Lorsque vous exécutez une traversée Gremlin qui renvoie un grand nombre de résultats, Neptune les renvoie au client par lots via la connexion. WebSocket Neptune envoie les lots de résultats au fur et à mesure qu'ils sont produits, sans attendre que le client en demande davantage. Cela peut être avantageux si vous souhaitez traiter les résultats tels qu'ils sont renvoyés par le serveur, mais cela nécessite d'utiliser des modèles d'itération différée pour éviter de collecter l'ensemble des résultats dans la mémoire.

Neptune envoie les résultats par lots de 64 par WebSocket image par défaut. Vous ne pouvez pas modifier cette valeur par défaut côté serveur, mais le client peut modifier la taille du lot au cas par cas à l'aide de l'option batchSizerequest (appelée Tokens.ARGS_BATCH_SIZE dans le pilote Java ou connectionPool.resultIterationBatchSize par défaut au niveau du pilote).

Pour plus de détails sur la configuration des pilotes batchSize dans d'autres langues, consultez la section Configuration de chaque pilote dans la documentation sur les pilotes et variantes d'Apache TinkerPop Gremlin.

Comme le serveur envoie automatiquement les résultats, la contre-pression côté client est gérée implicitement via le protocole TCP et le contrôle de flux. WebSocket Si le client met du temps à lire depuis le socket, les écritures du serveur finiront par être bloquées jusqu'à ce que le client rattrape son retard.

Important

Le streaming est particulièrement efficace avec les traversées qui peuvent produire des résultats de manière incrémentielle. Les traversées qui incluentorder(),groupCount(), group()dedup(), ou d'autres étapes nécessitant la réalisation de la traversée complète avant d'émettre des résultats permettront à Neptune de matérialiser l'ensemble de résultats en mémoire avant le début de la diffusion. Dans ces cas, le traitement par lots réduit tout de même la charge de sérialisation par image, mais ne réduit pas l'utilisation de la mémoire côté serveur.

Consommer les résultats de manière incrémentielle

Pour traiter les résultats au fur et à mesure qu'ils arrivent, itérez lentement en utilisanthasNext()/next()ou équivalent APIs plutôt que de rassembler tous les résultats dans une liste. Vous pouvez l'utiliser next(batchSize) pour extraire les résultats dans des lots au niveau de l'application, ce qui vous permet d'effectuer un travail intermédiaire entre les lots pendant que le serveur continue de produire des résultats.

Exemple Java (bytecode GLV)
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");
}
Exemple 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")
Exemple 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)
}
Exemple.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");
}
Exemple 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`);
}

Consommation rapide ou consommation incrémentielle

Le streaming vous permet de traiter les résultats de manière incrémentielle au fur et à mesure que des données supplémentaires sont récupérées et renvoyées. Les méthodes suivantes bloquent jusqu'à ce que la totalité du jeu de résultats soit collectée en mémoire, empêchant ainsi votre application d'agir sur les résultats lorsqu'ils arrivent :

  • Java : toList() ou toSet()

  • Python : toList() ou toSet()

  • Allez : ToList()ToSet(), ou GetResultSet().GetAll()

  • .NET : ToList() ou Promise()

  • Node.js : toList()

Note

Les données circulent toujours de manière incrémentielle sur la WebSocket connexion, même en utilisant ces méthodes. La différence est que votre demande ne peut pas traiter les résultats individuels tant que la collection complète n'est pas complète. Pour traiter les résultats au fur et à mesure qu'ils arrivent, utilisez les modèles d'itération différée ou de traitement par lots illustrés dans les exemples ci-dessus.