优化 BigQuery 资源的使用，使用 Google Dataflow 从 GCS 加载 200 万个 JSON 文件

Posted 2023-03-25

【中文标题】优化 BigQuery 资源的使用，使用 Google Dataflow 从 GCS 加载 200 万个 JSON 文件

【英文标题】：Optimize the use of BigQuery resources to load 2 million JSON files from GCS using Google Dataflow

【发布时间】：2020-06-12 02:59:30

【问题描述】：

我有一个庞大的数据库，其中包含约 240 万个 JSON 文件，这些文件本身包含多条记录。我创建了一个简单的 apache-beam 数据管道（如下所示），遵循以下步骤：

使用 glob 模式从 GCS 存储桶中读取数据。 从 JSON 数据中提取记录。 转换数据：将字典转换为 JSON 字符串、解析时间戳等。 写入 BigQuery。

# Pipeline
pipeline_options = PipelineOptions(pipeline_args)
pipeline_options.view_as(SetupOptions).save_main_session = save_main_session
p = beam.Pipeline(options=pipeline_options)

# Read
files = p | 'get_data' >> ReadFromText(files_pattern)

# Transform
output = (files
          | 'extract_records' >> beam.ParDo(ExtractRecordsFn())
          | 'transform_data' >> beam.ParDo(TransformDataFn()))

# Write
output | 'write_data' >> WriteToBigQuery(table=known_args.table,
                                         create_disposition=beam.io.BigQueryDisposition.CREATE_NEVER,
                                         write_disposition=beam.io.BigQueryDisposition.WRITE_EMPTY,
                                         insert_retry_strategy='RETRY_ON_TRANSIENT_ERROR',
                                         temp_file_format='NEWLINE_DELIMITED_JSON')

# Run
result = p.run()
result.wait_until_finish()

我已经用最小的样本数据集测试了这个管道，并且按预期工作。但我对 BigQuery 资源和配额的最佳使用持怀疑态度。 batch load quotas are very restrictive，由于要解析和加载大量文件，我想知道我是否遗漏了一些可以保证管道将遵守配额并以最佳方式运行的设置。我不想超过配额，因为我在同一个项目中运行 BigQuery 的其他负载。

我还没有完全理解 WriteToBigQuery() 转换的一些参数，特别是 batch_size、max_file_size 和 max_files_per_bundle，或者它们是否有助于优化 BigQuery 的加载作业。你能帮我解决这个问题吗？

更新

我不仅关心 BigQuery 配额，而且此管道使用的其他资源的 GCP 配额也是一个问题。

我尝试在目标数据（约 240 万个文件）上运行我的简单管道，但收到以下警告消息：

项目 [my-project] 的配额不足，无法使用区域 us-central1 中的 1 个实例执行此工作流。配额汇总（必需/可用）：1/16 个实例、1/16 个 CPU、250/2096 个磁盘 GB、0/500 SSD 磁盘 GB、1/99 个实例组、1/49 个托管实例组、1/99 个实例模板、 1/0 正在使用的 IP 地址。有关请求更多配额的信息，请参阅https://cloud.google.com/compute/docs/resource-quotas。

我不完全理解该消息。该过程成功激活了 8 个工作人员，并使用了 8 个可用的正在使用的 IP 地址中的 8 个。这是一个问题吗？我该如何解决？

【参考方案1】：

如果您担心加载作业配额，可以尝试将数据流式传输到带有较少 restrictive quota policy 的 bigquery。

要实现您想做的事情，您可以尝试 Google 提供的模板或直接参考他们的代码。

Cloud Storage Text to BigQuery (Stream) [code] Cloud Storage Text to BigQuery (Batch)

最后但并非最不重要的一点是，可以在Google BigQuery I/O connector 上找到更多详细信息。

【讨论】：

我将尝试数据流。我在想DataflowRunner 和 BigQuery 模块中有一些设置可以让我以最佳方式管理资源。