如何用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(...))

WaitPTransform 封装了这种模式。