google-cloud-dataflow - 数据流性能问题

Question

我知道几周前对 CDF 服务进行了更新(默认工作人员类型和附加的 PD 已更改)，并且很明显这会使批处理作业变慢。然而，我们的工作绩效已经下降，超出了实际满足我们业务需求的程度。

例如，特别是对于我们的一项作业:它从 BigQuery 中的表中读取约 270 万行，有 6 个辅助输入(BQ 表)，执行一些简单的字符串转换，最后将多个输出 (3) 写入大查询。过去，这需要 5-6 分钟，而现在则需要 15-20 分钟 - 无论我们使用多少个虚拟机。

我们可以做些什么来使速度恢复到我们以前看到的速度吗？

以下是一些统计数据:

1. 从包含 2,744,897 行 (294MB) 的 BQ 表中读取
2. 6 个 BQ 侧输入
3. 3 个多重输出到 BQ，其中 2 个为 2,744,897，其他为 1,500 行
4. 在区域 asia-east1-b 中运行
5. 以下时间包括工作池启动和关闭的时间

10 个虚拟机 (n1-standard-2)16分5秒2015-04-22_19_42_20-4740106543213058308

10 个虚拟机 (n1-standard-4)17分11秒2015-04-22_20_04_58-948224342106865432

10 个虚拟机 (n1-standard-1)18分44秒2015-04-22_19_42_20-4740106543213058308

20 个虚拟机 (n1-standard-2)22分53秒2015-04-22_21_26_53-18171886778433479315

50 个虚拟机 (n1-standard-2)17 分 26 秒2015-04-22_21_51_37-16026777746175810525

100 个虚拟机 (n1-standard-2)19分33秒2015-04-22_22_32_13-9727928405932256127

Answer 1

证据似乎表明您的管道处理侧面输入的方式存在问题。具体来说，对于主输入的每个元素，辅助输入很可能会一次又一次地从 BigQuery 重新读取。这与 Dataflow Worker 使用的虚拟机类型的更改完全正交，如下所述。

这与 Dataflow SDK for Java 版本 0.3.150326 中所做的更改密切相关。在此版本中，我们更改了侧输入 API 以应用于每个窗口。对 sideInput() 的调用现在仅返回与主输入元素的窗口相对应的特定窗口中的值，而不是整个侧输入 PCollectionView。因此，无法再从 DoFn 的 startBundle 和 finishBundle 调用 sideInput()，因为窗口不尚未可知。

例如，以下代码片段存在一个问题，会导致重新读取每个输入元素的侧面输入。

@Override
public void processElement(ProcessContext c) throws Exception {
  Iterable<String> uniqueIds = c.sideInput(iterableView);

  for (String item : uniqueIds) {
    [...]
  }

  c.output([...]);
}

可以通过在第一次调用 processElement 期间将侧输入缓存到转换的 List 成员变量(假设它适合内存)来改进此代码，并且在后续调用中使用缓存的 List 而不是侧面输入。

此解决方法应该可以恢复您之前看到的性能，当时可以从 startBundle 调用侧面输入。从长远来看，我们将致力于更好地缓存侧面输入。 (如果这不能帮助完全解决问题，请通过电子邮件与我们联系并分享相关的代码片段。)

<小时/>

另外，确实在 2015 年 4 月 9 日左右对 Cloud Dataflow Service 进行了更新，更改了 Dataflow Worker 使用的虚拟机的默认类型。具体来说，我们减少了每个工作线程的默认核心数量，因为我们的基准测试表明它对于典型作业来说具有成本效益。这并不是任何类型的数据流服务的减慢——默认情况下，它只是以每个工作线程更少的资源运行。用户仍然可以选择覆盖工作人员数量以及工作人员使用的虚拟机类型。