View a markdown version of this page

使用 Gremlin 流式传输查询结果 - Amazon Neptune
逐步消耗结果急切消费与增量消费
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异，请参阅 中国的 Amazon Web Services 服务入门 (PDF)。

本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

使用 Gremlin 流式传输查询结果

当您运行返回大量结果的 Gremlin 遍历时，Neptune 会通过连接将它们分批传输回客户端。 WebSocket Neptune 在生成结果批次时发送结果批次，无需等待客户端请求更多。如果您想在服务器返回结果时对其进行处理，这可能非常有利，但需要使用延迟迭代模式以避免将完整的结果集收集到内存中。

默认情况下，海王星分批发送结果，每 WebSocket 帧 64 个。您无法更改此服务器端默认值，但是可以使用请求选项（在 Java 驱动程序中调Tokens.ARGS_BATCH_SIZE用，或connectionPool.resultIterationBatchSize作为驱动程序级别的默认值），根据客户端的每个batchSize请求来覆盖批处理大小。

有关使用其他语言驱动程序进行配置batchSize的详细信息，请参阅 Apache TinkerPop Gremlin 驱动程序和变体文档中每个驱动程序的配置部分。

由于服务器会自动推送结果，因此客户端的背压是通过 TCP 和流量控制隐式处理的。 WebSocket 如果客户端从套接字读取的速度很慢，服务器的写入操作最终将阻塞，直到客户端赶上。

重要

通过遍历可以逐步产生结果，流式传输最为有效。包含order()、、groupCount()group()dedup()、或其他需要在发出结果之前完成完整遍历的步骤的遍历将导致 Neptune 在开始流式传输之前在内存中实现整个结果集。在这些情况下，批处理仍然可以减少每帧的序列化开销，但不会减少服务器端的内存使用量。

逐步消耗结果

要在结果到达时对其进行处理，请使用hasNext()/next()或等效的 API 进行延迟迭代，而不是将所有结果收集到列表中。您可以使用next(batchSize)在应用程序级批次中提取结果，从而允许您在服务器继续生成结果的同时，在批次之间执行中间工作。

例 Java（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");
}
例 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")
例 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)
}
例.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");
}
例 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`);
}

急切消费与增量消费

流式传输允许您在获取和返回更多数据时逐步处理结果。在将整个结果集收集到内存中之前，以下方法会阻塞，从而防止您的应用程序在结果到达时对结果采取行动：

  • Java：toList()或 toSet()

  • Python：toList()或 toSet()

  • 去:ToList()、ToSet()、或 GetResultSet().GetAll()

  • .NET：ToList()或 Promise()

  • Node.js: toList()

注意

即使使用这些方法，数据仍会以增量方式通过 WebSocket 连接流动。区别在于，在整个收集完成之前，您的应用程序无法处理单个结果。要在结果到达时对其进行处理，请使用上面示例中显示的延迟迭代或批处理模式。