java - 如何用google cloud dataflow和java依次生成gcs文件？

Question

我有一个以一个 gcs 文件作为输入并生成两个 gcs 输出文件的管道。
一个输出文件包含错误信息，另一个包含正常信息。
我有一个云函数，带有两个输出文件的 gcs 触发器。
只有当错误信息文件为 0 字节时，我才想对正常信息文件做一些事情。
所以我必须让错误信息文件比正常信息文件更早生成，以检查错误信息文件的大小。

现在我使用2个TextIO.Write来生成这两个文件。
但是我无法控制先生成哪个。
在云函数中，我让普通信息文件通过重试检查错误信息文件的大小。
但是云功能作为 540 秒的超时限制，所以在生成错误信息文件之前我无法重试。
我如何在 Cloud Dataflow 中处理此问题？
我可以通过编程方式在正常信息文件之前生成错误信息文件吗？

Answer 1

您可以通过使用侧输入来完成这样的排序。例如，

error_pcoll = ...
good_data_pcoll = ...

error_write_result = error_pcoll | beam.io.WriteToText(...)
(good_data_pcoll
 | beam.Map(
       # This lambda simply emits what it was given.
       lambda element, blocking_side: element,
       # This side input isn't used,
       # but will force error_write_result to be computed first.
       blocking_side=beam.pvalue.AsIterable(error_write_result))
 | beam.io.WriteToText(...))

Wait PTransform 封装了这种模式。