【发布时间】：2020-03-26 19:40:23
【问题描述】：
我正在开发一个从 Google Cloud Storage (GCS) 目录中读取约 500 万个文件的管道。我已将其配置为在 Google Cloud Dataflow 上运行。
问题是当我启动管道时，“计算”所有文件的大小需要几个小时：
INFO:apache_beam.io.gcp.gcsio:Starting the size estimation of the input
INFO:apache_beam.io.gcp.gcsio:Finished computing size of: 10000 files
[...]
INFO:apache_beam.io.gcp.gcsio:Finished computing size of: 5480000 files
INFO:apache_beam.io.gcp.gcsio:Finished listing 5483720 files in 5549.38778591156 seconds.
INFO:apache_beam.io.gcp.gcsio:Starting the size estimation of the input
INFO:apache_beam.io.gcp.gcsio:Finished computing size of: 10000 files
[...]
INFO:apache_beam.io.gcp.gcsio:Finished computing size of: 5480000 files
INFO:apache_beam.io.gcp.gcsio:Finished listing 5483720 files in 7563.196493148804 seconds.
INFO:apache_beam.io.gcp.gcsio:Starting the size estimation of the input
INFO:apache_beam.io.gcp.gcsio:Finished computing size of: 10000 files
[...]
如您所见，计算大约 550 万个文件的大小需要一个半小时（5549 秒），然后又从头开始！第二遍又跑了2个小时，然后又开始了第三遍！截至撰写本文时，该作业在 Dataflow 控制台中仍然不可用，这让我相信这一切都发生在我的本地机器上，并且没有利用任何分布式计算。
当我使用较小的输入数据集（2 个文件）测试管道时，它会重复大小估计 4 次：
INFO:apache_beam.io.gcp.gcsio:Starting the size estimation of the input
INFO:apache_beam.io.gcp.gcsio:Finished listing 2 files in 0.33771586418151855 seconds.
INFO:apache_beam.io.gcp.gcsio:Starting the size estimation of the input
INFO:apache_beam.io.gcp.gcsio:Finished listing 2 files in 0.1244659423828125 seconds.
INFO:apache_beam.io.gcp.gcsio:Starting the size estimation of the input
INFO:apache_beam.io.gcp.gcsio:Finished listing 2 files in 0.13422417640686035 seconds.
INFO:apache_beam.io.gcp.gcsio:Starting the size estimation of the input
INFO:apache_beam.io.gcp.gcsio:Finished listing 2 files in 0.14139890670776367 seconds.
按照这个速度，在 Dataflow 作业开始之前，仅对所有 5.5M 文件执行 4 次 GCS 大小估计就需要大约 8 小时。
我的管道配置了 --runner=DataflowRunner 选项，因此它应该在 Dataflow 中运行：
python bigquery_import.py --runner=DataflowRunner #other options...
管道从 GCS 读取如下：
parser = argparse.ArgumentParser()
parser.add_argument(
'--input',
required=True,
help='Input Cloud Storage directory to process.')
known_args, pipeline_args = parser.parse_known_args(argv)
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = True
with beam.Pipeline(options=pipeline_options) as p:
files = p | beam.io.ReadFromText('gs://project/dir/*.har.gz')
完整代码请参考 GitHub 上的bigquery_import.py。
我很困惑为什么这个繁琐的过程会发生在 Dataflow 环境之外，以及为什么需要多次执行。我是从 GCS 正确读取文件还是有更有效的方法？
【问题讨论】：
-
FWIW 我正在更新不存在此问题的旧 Dataflow Java SDK 实现。源代码供参考：github.com/HTTPArchive/bigquery/blob/master/dataflow/java/src/…
标签： python google-cloud-dataflow apache-beam