数据流作业GCS到发布/订阅的最大批处理大小

如何解决数据流作业GCS到发布/订阅的最大批处理大小

我正在使用默认数据流模板GCS到Pub / Sub。在云存储中输入文件，文件大小为300MB，每个文件有2-3百万行。

启动数据流批处理作业时，出现以下错误

来自工作程序的错误消息：javax.naming.SizeLimitExceededException：发布/订阅消息大小（1089680070）超出了最大批处理大小（7500000） org.apache.beam.sdk.io.gcp.pubsub.PubsubIO $ Write $ PubsubBoundedWriter.processElement（PubsubIO.java:1160）

来自文档：批量发布/订阅最多可接收1,000条消息，并且批量大小不能超过10 MB。

这是否意味着我必须将输入文件拆分为10MB块或1000条消息才能发布？

什么是将如此大的文件（每个300MB）加载到pubsub的推荐方法？

预先感谢您的帮助。

解决方法

这是数据流方面的已知限制，此刻存在feature request以增加批大小的大小。使用+1按钮并为问题加注星标，以跟踪进展情况。

我建议您在建议解决方法的地方检查此post。重要的是要考虑到这种解决方法意味着对Cloud Storage Text to Pub/Sub模板的修改以实现此处提到的自定义转换。

另一方面，您可以尝试创建云功能来拆分文件，然后再由Dataflow处理，我想是这样的：

创建一个“分段”存储桶以上传大文件。
写一个Cloud Function来分割文件，并将小的块写入另一个存储桶中。您可以尝试使用filesplit Python软件包来做到这一点。
每次使用Google Cloud Storage Triggers在“登台”存储桶中上传新文件时，都将触发“云功能”运行。
一旦文件分成小块，请使用相同的Cloud Function从“ staging”存储桶中删除大文件，以免产生额外费用。
使用Dataflow模板的Cloud Storage Text进行发布/订阅以处理第二个存储桶中的小块。