google-cloud-platform - BigQueryIO - 流和 FILE_LOADS 的写入性能

Question

我的管道:Kafka -> 数据流流 (Beam v2.3) -> BigQuery

鉴于低延迟对我来说并不重要，我使用 FILE_LOADS降低成本，像这样:

BigQueryIO.writeTableRows()
  .withJsonSchema(schema)
  .withWriteDisposition(WriteDisposition.WRITE_APPEND)
  .withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
  .withMethod(Method.FILE_LOADS)
  .withTriggeringFrequency(triggeringFrequency)      
  .withCustomGcsTempLocation(gcsTempLocation)
  .withNumFileShards(numFileShards) 
  .withoutValidation()
  .to(new SerializableFunction[ValueInSingleWindow[TableRow], TableDestination]() {
    def apply(element: ValueInSingleWindow[TableRow]): TableDestination = {
      ...
    }
  }

这个数据流步骤在管道中引入了更大的延迟，因此它无法跟上 Kafka 吞吐量(小于 50k 事件/秒)，即使是 40 n1-standard-s4 worker 。如下面的屏幕截图所示，这一步的系统延迟非常大(接近管道正常运行时间)，而 Kafka 系统延迟只有几秒钟。



如果我理解正确，Dataflow 会将元素写入 gcsTempLocation 中的 numFileShards，并且每个触发频率都会启动加载作业以将它们插入 BigQuery。例如，如果我选择 5 分钟的触发频率，我可以看到(使用 bq ls -a -j )所有加载作业需要不到 1 分钟才能完成。但是这一步仍然引入了越来越多的延迟，导致 Kafka 消耗的元素越来越少(感谢 bcackpressure)。增加/减少 numFileShards 和 triggeringFrequency 不能解决问题。

我没有手动指定任何窗口，我只是默认一个。文件未在 gcsTempLocation 中累积。

知道这里出了什么问题吗？

Answer 1

您提到您没有明确指定窗口，这意味着默认情况下 Dataflow 将使用“全局窗口”。 windowing documentation包含此警告:

Caution: Dataflow's default windowing behavior is to assign all elements of a PCollection to a single, global window, even for unbounded PCollections. Before you use a grouping transform such as GroupByKey on an unbounded PCollection, you must set a non-global windowing function. See Setting Your PCollection's Windowing Function.

If you don't set a non-global windowing function for your unbounded PCollection and subsequently use a grouping transform such as GroupByKey or Combine, your pipeline will generate an error upon construction and your Dataflow job will fail.

You can alternatively set a non-default Trigger for a PCollection to allow the global window to emit "early" results under some other conditions.

您的管道似乎没有进行任何显式分组，但我想知道通过 BigQuery 写入的内部分组是否会导致问题。

如果您的下游 DropInputs，您能在 UI 中看到吗？已经收到任何元素？如果没有，这表明数据在上游 BigQuery 步骤中被阻塞。