在Apache Beam/Dataflow Python流中写入文本文件

Question

我有一个非常基本的Python Dataflow工作，从pub/sub读取一些数据，应用FixedWindow并写入Google Cloud Storage。

transformed = ...
transformed | beam.io.WriteToText(known_args.output)

输出被写入--output中特定的位置，但只写入临时阶段，即。

gs://MY_BUCKET/MY_DIR/beam-temp-2a5c0e1eec1c11e8b98342010a800004/...some_UUID...

当进一步测试时，我注意到streaming_wordcount示例也有同样的问题，但是标准wordcount示例写得很好。也许问题在于开窗，或者从PubSub阅读？

WriteToText似乎与PubSub的流源不兼容。可能有变通办法，或者Java版本可能兼容，但我选择使用完全不同的解决方案。

Answer 1

Python SDK中的WriteToText转换不支持流。

相反，您可以考虑apache_beam.io.fileio中的转换。在本例中，您可以编写如下内容（假设窗口为10分钟）：

my_pcollection = (p | ReadFromPubSub(....)
                    |  WindowInto(FixedWindows(10*60))
                    |  fileio.WriteToFiles(path=known_args.output))

这足以为每个窗口写出单独的文件，并随着流的前进继续这样做。

您将看到如下所示的文件（假设输出为gs://mybucket/)。当windows被触发时，文件将被打印出来：

gs://mybucket/output-1970-01-01T00_00_00-1970-01-01T00_10_00-0000-00002
gs://mybucket/output-1970-01-01T00_00_00-1970-01-01T00_10_00-0001-00002
gs://mybucket/output-1970-01-01T00_10_00-1970-01-01T00_20_00-0000-00002
gs://mybucket/output-1970-01-01T00_10_00-1970-01-01T00_20_00-0001-00002
...

默认情况下，这些文件具有$prefix-$start-$end-$pane-$shard-of-$numshards$suffix$compressionsuffix名称-其中prefix默认情况下是output，但是您可以传递一个更复杂的函数来命名文件。

如果您希望自定义文件的编写方式（例如文件的命名、数据的格式或类似的内容），可以查看WriteTofiles中的额外参数。

您可以在这里看到一个在Beam测试中使用的转换的示例，其中包含了更复杂的参数--但听起来默认行为对您来说应该足够了。