Google Dataflow batch file processing poor performance

Question

I'm trying to build a pipeline using Apache Beam 2.16.0 for processing large amount of XML files. Average count is seventy million per 24 hrs, and at peak load it can go up to half a billion. File sizes varies from ~1 kb to 200 kb (sometimes it can be even bigger, for example 30 mb)

File goes through various transformations and final destination is BigQuery table for further analysis. So, first I read xml file, then deserialize into POJO (with help of Jackson) and then apply all required transformations. Transformations works pretty fast, on my machine I was able to get about 40000 transformations per second, depending on file size.

My main concern is file reading speed. I have feeling that all reading is done only via one worker, and I don't understand how this can be paralleled. I tested on 10k test files dataset.

Batch job on my local machine (MacBook pro 2018: ssd, 16 gb ram and 6-core i7 cpu) can parse about 750 files/sec. If I run this on DataFlow, using n1-standard-4 machine, I can get only about 75 files/sec. It usually doesn't scale up, but even if it does (sometimes up to 15 workers), I can get only about 350 files/sec.

More interesting is streaming job. It immediately starts from 6-7 workers and on UI I can see 1200-1500 elements/sec, but usually it doesn't show speed, and if I select last item on page, it shows that it already processed 10000 elements.

The only difference between batch and stream job is this option for FileIO:

.continuously(Duration.standardSeconds(10), Watch.Growth.never()))

Why this makes such a big difference in processing speed?

Run parameters:

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

Run region and bucket region are the same.

Pipeline:

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()));

Example of xml file:

<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>

Real life file and project are much more complex, with lots of nested nodes and huge amount of transformation rules.

Simplified code on GitHub: https://github.com/costello-art/dataflow-file-io It contains only "bottleneck" part - reading files and deserializing into POJO.

If I can process about 750 files/sec on my machine (which is one powerful worker), then I expect to have about 7500 files/sec on similar 10 workers in Dataflow.

Interesting... Bundle sizes are in general larger in batch vs streaming mode, but not sure if this is related. Could you try making use of this, its depricated, but this is just to check something : Reshuffle.ViaRandomKey() between the FileIO.match() and your XmlToPojoDoFn() — Reza Rokni
@RezaRokni I didn't notice any difference. But during testing I changed read operation: FileIO.Match -> FileIO.ReadMatches -> read file as bytes -> convert bytes to POJO. I also did testing on my production code and on larger dataset (144k and 1m files) in batch mode: on n1-standard-2 I was able to get about 1000k files/sec with 17 workers. This is much better, but still I'm not close to 850 files/sec like on my local machine. Need to do more testing. I updated github code with this approach. — costello
Btw, when you are doing testing on your local machine, where are the files sitting? On your local disk or in a cloud bucket? — Reza Rokni
@RezaRokni locally (ssd). I wonder if I can achieve similar read performance on cloud — costello
Your files are sitting in Cloud Storage I assume, they will be pulled onto the workers. — Reza Rokni

Gonzalo Pérez Fernández Gonzalo Pérez Fernández · Accepted Answer · 2020-01-14T15:44:15

I was trying to make a test code with some functionality to check the behavior of the FileIO.match and the number of workers [1].

In this code I set the value numWorkers to 50, but you can set the value you need. What I could see is that the FileIO.match method will find all the links that match these patterns but after that, you must deal with the content of each file separately.

For example, in my case I created a method that receives each file and then I divided the content by "new_line (\n)" character (but here you can handle it as you want, it depends also on the type of file, csv, xml, ...).

Therefore, I transformed each line to TableRow, format that BigQuery understands, and return each value separately (out.output(tab)), this way, Dataflow will handle the lines in different workers depending the workload of the pipeline, for example 3000 lines in 3 different workers, each one with 1000 lines.

At the end, since it is a batch process, Dataflow will wait to process all the lines and then insert this in BigQuery.

I hope this test code helps you with yours.

[1] https://github.com/GonzaloPF/dataflow-pipeline/blob/master/java/randomDataToBQ/src/main/fromListFilestoBQ.java

Google Dataflow batch file processing poor performance

1 Answers