Google Dataflow批处理文件性能差

Question

我试图使用Apache Beam2.16.0构建一个流水线来处理大量的XML文件。平均每24小时的计数是7000万，在高峰负载时，它可以上升到5亿。文件大小从1 kb到200 kb不等（有时甚至更大，例如30 mb）

文件经过各种转换，最终目标是BigQuery表，以便进一步分析。因此，我首先读取xml文件，然后反序列化为POJO（在Jackson的帮助下），然后应用所有所需的转换。转换工作得非常快，在我的机器上，根据文件大小，我每秒可以得到大约40000个转换。

我主要关心的是文件读取速度。我感觉所有的阅读都是通过一个工人完成的，我不明白这怎么能并行不悖。我在10K测试文件数据集上进行了测试。

为什么这会在处理速度上产生如此大的差异？

运行参数：

--runner=DataflowRunner
--project=<...>
--inputFilePattern=gs://java/log_entry/*.xml
--workerMachineType=n1-standard-4
--tempLocation=gs://java/temp
--maxNumWorkers=100

运行区域和桶区域相同。

pipeline.apply(
  FileIO.match()
    .withEmptyMatchTreatment(EmptyMatchTreatment.ALLOW)
    .filepattern(options.getInputFilePattern())
    .continuously(Duration.standardSeconds(10), Watch.Growth.never()))
  .apply("xml to POJO", ParDo.of(new XmlToPojoDoFn()));

<LogEntry><EntryId>0</EntryId>
    <LogValue>Test</LogValue>
    <LogTime>12-12-2019</LogTime>
    <LogProperty>1</LogProperty>
    <LogProperty>2</LogProperty>
    <LogProperty>3</LogProperty>
    <LogProperty>4</LogProperty>
    <LogProperty>5</LogProperty>
</LogEntry>

如果我可以在我的机器上处理大约750个文件/秒（这是一个强大的worker），那么我希望在DataFlow中类似的10个worker上处理大约7500个文件/秒。

Answer 1

我试图创建一个带有某些功能的测试代码，以检查fileio.match的行为和工作人员的数量[1]。

在这段代码中，我将值numWorkers设置为50，但是您可以设置您需要的值。我可以看到的是，fileio.match方法将查找与这些模式匹配的所有链接，但在此之后，您必须分别处理每个文件的内容。

例如，在我的例子中，我创建了一个接收每个文件的方法，然后用“new_line(\n)”字符对内容进行分割（但是在这里您可以根据需要来处理它，这也取决于文件的类型，csv、xml、...)。

因此，我将每一行转换为tableRow，即BigQuery理解的格式，并分别返回每个值(out.output(tab))，这样，Dataflow将根据管道的工作负载在不同的worker中处理这些行，例如，在3个不同的worker中处理3000行，每个worker有1000行。

最后，由于这是一个批处理过程，Dataflow将等待处理所有行，然后将其插入到BigQuery中。

我希望这段测试代码对您的测试有所帮助。