在 Python 中的数据流中写入到云存储的动态目标答案

【问题标题】：Write to dynamic destination to cloud storage in dataflow in Python在 Python 中的数据流中写入到云存储的动态目标
【发布时间】：2018-02-15 17:04:07
【问题描述】：

我试图从云存储中的一个大文件中读取数据，并根据给定的字段对它们进行分片。

我打算阅读 |映射（lambda x: (x[key field], x)) | GroupByKey |使用关键字段的名称写入文件。

但是我找不到动态写入云存储的方法。是否支持此功能？

谢谢你，宜清

【问题讨论】：

标签： python-2.7 google-cloud-storage google-cloud-dataflow apache-beam apache-beam-io

【解决方案1】：

是的，您可以使用FileSystems API 创建文件。

【讨论】：

谢谢！我在 ParDo 中使用 FileSystems.create 句柄来编写分组结果。然而，GroupByKey 似乎会等待所有数据被读取，然后开始写入单个文件。所以我有两个后续问题：1）我可以使用 FileSystems API 使用通配符吗？ 2）有没有办法让GroupBy不必等待所有数据，否则可能会出现内存问题。再次感谢！

【解决方案2】：

2.14.0 的 Beam python SDK 中添加了一个实验性写入，beam.io.fileio.WriteToFiles：

my_pcollection | beam.io.fileio.WriteToFiles(
      path='/my/file/path',
      destination=lambda record: 'avro' if record['type'] == 'A' else 'csv',
      sink=lambda dest: AvroSink() if dest == 'avro' else CsvSink(),
      file_naming=beam.io.fileio.destination_prefix_naming())

它可用于每条记录写入不同的文件。

您可以跳过GroupByKey，只需使用destination 来决定每条记录写入哪个文件。 destination的返回值需要是可以分组的值。

还有这里的 JIRA 问题：

https://issues.apache.org/jira/browse/BEAM-2857

【讨论】：