如何在 R 中执行并行异步 POST API 调用？

Question

我在 AWS Lambda 上创建了一个 API 端点，我使用 POST 请求（带有 JSON 数据输入）调用它以获取结果作为响应（作为 JSON 数据输出）。

现在我有 100 万个需要使用 API 处理的 data.table。一个 API 执行大约需要 600 毫秒，AWS Lambda 每秒最多允许 3000 个请求。为了更快地执行，我想使用系统中的所有 64 个内核来生成 64 个并发请求，但我不希望这些请求等到它们得到响应（结果）之后再生成下一个 64 个请求等等（基本上我想要达到 3000 个请求阈值）。为此，我需要异步生成请求，每个请求不等待它的响应，并且一旦生成响应以将其附加到大列表或 data.table

我查看了 curl、Rcurl、future 和 doFuture 包文档，但无法找到找到解决这个问题的任何东西。任何帮助将不胜感激

我在this post 上发现了一个类似的问题，但答案并不完整。

在异步 foreach（或类似功能的函数）中运行的示例伪代码：

output = foreach(i = 1:n) %dopar%
{
  x = input[i]
  body = toJSON(x)
  url = "https://exampleURL.amazonaws.com/dev/LambdaTest"
  response = as.data.table(fromJSON(content(POST(url,
                                                 body = body,
                                                 content_type_json()))))
  return(response)
}

这里的 data 是 data.table，其中包含需要单独传递的 n 个子集。

Answer 1

您可以使用crul 包。它有两个异步接口，Async 用于处理相同的许多 URL，AsyncVaried 可以在任何配置中构造 HTTP 请求，然后将它们传递给 AsyncVaried 以处理这些上的异步请求

library(crul)

形成请求

req1 <- HttpRequest$new(
  url = "https://httpbin.org/post", 
  headers = list(`Content-Type` = "application/json")
)$post(body = jsonlite::toJSON(iris[1,]))
req2 <- HttpRequest$new(
  url = "https://httpbin.org/post", 
  headers = list(`Content-Type` = "application/json")
)$post(body = jsonlite::toJSON(iris[2,]))

创建一个 AsyncVaried 对象

out <- AsyncVaried$new(req1, req2)

做请求

out$request()

获取状态码、标头等

out$status_code()

获取 json 响应并解析为 R 列表

lapply(out$parse(), jsonlite::fromJSON)