google-cloud-datastore - 如何加快批量导入到多个工作人员的谷歌云数据存储？

Question

我有一个基于 apache-beam 的数据流作业要使用 vcf source 读取从单个文本文件(存储在谷歌云存储中)，将文本行转换为数据存储Entities并将它们写入datastore sink .工作流程工作正常，但我注意到的缺点是:

• 写入数据存储区的速度最多为每秒 25-30 个实体。
• 我尝试使用 --autoscalingAlgorithm=THROUGHPUT_BASED --numWorkers=10 --maxNumWorkers=100 但执行似乎更喜欢一个 worker (见下图:目标 worker 曾经增加到 2 但“基于在当前运行的步骤中并行化工作的能力”减少到 1)。

我没有为键使用祖先路径；所有实体都是相同的 kind。

管道代码如下所示:

def write_to_datastore(project, user_options, pipeline_options):
"""Creates a pipeline that writes entities to Cloud Datastore."""
  with beam.Pipeline(options=pipeline_options) as p:
  (p
   | 'Read vcf files' >> vcfio.ReadFromVcf(user_options.input)
   | 'Create my entity' >> beam.ParDo(
     ToEntityFn(), user_options.kind)
   | 'Write to datastore' >> WriteToDatastore(project))

因为我有数百万行要写入数据存储，以每秒 30 个实体的速度写入会花费太长时间。

问题:输入只是一个巨大的压缩文件。是否需要拆分成多个小文件来触发多个worker？有没有其他方法可以加快导入速度？我是否遗漏了 num_workers 设置中的某些内容？谢谢!

Answer 1

我对 apache beam 不熟悉，答案是从一般流程的角度来看。

假设各个输入文件部分中的实体数据之间没有依赖关系，那么是的，使用多个输入文件肯定会有所帮助，因为所有这些文件都可以虚拟地并行处理(当然，取决于最大可用 worker 的数量)。

您可能不需要事先拆分巨大的 zip 文件，可能只需将单个输入数据流的段交给单独的数据段 worker 进行写入，如果这种切换的开销与实际的数据段处理相比，它本身可以忽略不计。

整体性能限制将是读取输入数据、将其拆分为段并移交给段数据 worker 的速度。

数据段 worker 会将它接收到的数据段进一步拆分为更小的 block ，最多相当于最多 500 个实体，这些实体可以在单个批处理操作中转换为实体并写入数据存储。根据所使用的数据存储客户端库，可以异步执行此操作，从而允许继续拆分为 block 并转换为实体，而无需等待先前的数据存储写入完成。

数据段 worker 的性能限制将是数据段可以拆分为 block 并将 block 转换为实体的速度

如果异步操作不可用或需要更高的吞吐量，则可以将每个 block 再次移交给段工作程序，由段工作程序执行到实体的转换和数据存储批量写入。

数据段 worker 级别的性能限制将只是数据段可以拆分成 block 并移交给 block worker 的速度。

通过这种方法，实际转换为实体并将它们批量写入数据存储(异步或非异步)将不再位于拆分输入数据流的关键路径中，我相信这是您当前的性能限制方法。