数据流性能问题答案

【问题标题】：Dataflow performance issues数据流性能问题
【发布时间】：2015-04-23 06:00:27
【问题描述】：

我知道几周前对 CDF 服务进行了更新（默认工作器类型和附加的 PD 已更改），并且很明显它会使批处理作业变慢。然而，我们的工作表现已经下降到无法真正满足我们的业务需求。

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

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

以下是一些统计数据：

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

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

【问题讨论】：

我查看了其中一项作业的步骤执行日志，似乎大部分时间（大约 17 分钟中的 9 分钟）已用于导入已由步骤写入的数据进入 BQ。我们将研究为什么这个导入过程变得如此缓慢。
在您弄清楚为什么它变得如此缓慢之前，我可以使用一种解决方法吗？
一位队友建议，速度缓慢可能是由于新 SDK 处理侧输入的方式发生了变化 - 请您查看 stackoverflow.com/questions/29718820/… 并检查它是否与您的工作相关？
另外，请参阅stackoverflow.com/questions/29685886/… 了解调试 BiqQuery 导出作业的提示 - 也许这些统计信息对您的性能调试很有用。如果这些都没有帮助，请告诉我。
@jkff WRT 到第一个链接，我们在第一次调用 processElement() 时缓存侧输入。我们过去是从 startBundle() 开始的，但当然 API 在上一个版本中发生了变化，所以我们将它移到了 processElement()。

标签： google-cloud-dataflow

【解决方案1】：

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

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

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

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

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

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

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

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

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

【讨论】：

这就是我们已经做的——我们在 processElement() 中读取一次侧输入，并将结果缓存在一个内部变量中（在我们的例子中是一个 HashMap），并将其用于对 processElement 的每个后续调用()。在最新版本之前，一切正常。
不，不会为每次调用 processElement() 创建一个 ParDo 的新实例。那将破坏捆绑的目的。然而，你的经验证据似乎暗示了一些类似的东西。我将单独与您联系，等我们查明真相后再发布答案。

【解决方案2】：

我们追查了这个问题。当侧输入从 BigQuery 表中读取数据时，该表已将其数据流入，而不是批量加载。当我们复制表格并从副本中读取时，一切正常。

但是，这只是一种解决方法。 Dataflow 应该能够将 BigQuery 中的流式表作为辅助输入来处理。

【讨论】：