如何在 Apache Beam 步骤中并行化 HTTP 请求？

Question

我有一个在 Google Dataflow 上运行的 Apache Beam 管道，其工作相当简单：

• 它从 Pub/Sub 读取单个 JSON 对象
• 解析它们
• 并通过 HTTP 将它们发送到某些 API

这个 API 要求我分批发送 75 个项目。所以我构建了一个 DoFn，它将事件累积在一个列表中，一旦我得到 75 个就通过这个 API 发布它们。结果太慢了，所以我而不是使用线程池在不同的线程中执行这些 HTTP 请求。

我现在的实现是这样的：

private class WriteFn : DoFn<TheEvent, Void>() {
  @Transient var api: TheApi

  @Transient var currentBatch: MutableList<TheEvent>

  @Transient var executor: ExecutorService

  @Setup
  fun setup() {
    api = buildApi()
    executor = Executors.newCachedThreadPool()
  }

  @StartBundle
  fun startBundle() {
    currentBatch = mutableListOf()
  }

  @ProcessElement
  fun processElement(processContext: ProcessContext) {
    val record = processContext.element()

    currentBatch.add(record)

    if (currentBatch.size >= 75) {
      flush()
    }
  }

  private fun flush() {
    val payloadTrack = currentBatch.toList()
    executor.submit {
      api.sendToApi(payloadTrack)
    }
    currentBatch.clear()
  }

  @FinishBundle
  fun finishBundle() {
    if (currentBatch.isNotEmpty()) {
      flush()
    }
  }

  @Teardown
  fun teardown() {
    executor.shutdown()
    executor.awaitTermination(30, TimeUnit.SECONDS)
  }
}

从数据进入 API 的意义上说，这似乎“很好”。但我不知道这是否是正确的方法，我感觉这很慢。

我认为它很慢的原因是在负载测试时（通过向 Pub/Sub 发送几百万个事件），管道将这些消息转发到 API（有响应时间不到 8 毫秒），而不是让我的笔记本电脑将它们输入 Pub/Sub。

我的实现有什么问题吗？这是我应该这样做的方式吗？

另外...我是否需要在我的 @FinishBundle 方法中等待所有请求完成（即通过获取执行程序返回的期货并等待它们）？

Answer 1

这里有两个相互关联的问题：

1. 你做对了吗/你需要改变什么吗？
2. 您需要在@FinishBundle 等待吗？

第二个答案：是的。但实际上你需要更彻底地冲洗，这一点将变得清晰。

一旦您的@FinishBundle 方法成功，Beam runner 将假定捆绑已成功完成。但是您的 @FinishBundle 仅发送请求 - 它并不能确保它们已成功。因此，如果请求随后失败，您可能会以这种方式丢失数据。您的@FinishBundle 方法实际上应该被阻塞并等待来自TheApi 的成功确认。顺便说一句，以上所有内容都应该是幂等的，因为在完成捆绑后，地震可能会发生并导致重试；-)

所以回答第一个问题：你应该改变什么吗？就如上。只要您确定在提交捆绑包之前提交了结果，以这种方式批处理请求的做法就可以工作。

您可能会发现这样做会导致您的管道变慢，因为@FinishBundle 比@Setup 更频繁地发生。要跨包批量处理请求，您需要使用状态和计时器的较低级别功能。我在https://beam.apache.org/blog/2017/08/28/timely-processing.html 写了一个人为的用例版本。我会对它对您的工作方式非常感兴趣。

当您的管道中存在持久的随机播放时，您所期望的极低延迟（在低毫秒范围内）可能只是不可用。