Dataflow performance issues

Question

I'm aware that an update was made to the CDF service a few weeks ago (default worker type & attached PD were changed), and it was made clear that it would make batch jobs slower. However, the performance of our jobs has degraded beyond the point of them actually fulfilling our business needs.

For example, for one of our jobs in particular: it reads ~2.7 million rows from a table in BigQuery, has 6 side inputs (BQ tables), does some simple String transformations, and finally writes multiple outputs (3) to BigQuery. This used to take 5-6 minutes and now it takes anywhere between 15-20 mins - not matter how many VM's we chuck at it.

Is there anything we can do to get the speeds back up to what we used to see?

Here are some stats:

Reading from a BQ table with 2,744,897 rows (294MB)
6 BQ side inputs
3 multi-outputs to BQ, 2 of which are 2,744,897 and the other 1,500 rows
Running in zone asia-east1-b
Times below include worker pool spin up and tear down

10 VMs (n1-standard-2) 16 mins 5 sec 2015-04-22_19_42_20-4740106543213058308

10 VMs (n1-standard-4) 17 min 11 sec 2015-04-22_20_04_58-948224342106865432

10 VMs (n1-standard-1) 18 min 44 sec 2015-04-22_19_42_20-4740106543213058308

20 VMs (n1-standard-2) 22 min 53 sec 2015-04-22_21_26_53-18171886778433479315

50 VMs (n1-standard-2) 17 min 26 sec 2015-04-22_21_51_37-16026777746175810525

100 VMs (n1-standard-2) 19 min 33 sec 2015-04-22_22_32_13-9727928405932256127

I took a look at the step execution logs of one of these jobs and it appears that the bulk of time (about 9 minutes out of 17) has been spent on importing the data already written by the step into BQ. We'll be looking into why this import process has become so slow. — jkff
is there a workaround I could use until you figure out why it's become so slow? — Graham Polley
A teammate suggested that the slowness may be explained by a change in the way the new SDK treats side inputs - can you please take a look at stackoverflow.com/questions/29718820/… and check if it's relevant to your job? — jkff
Also, see stackoverflow.com/questions/29685886/… for tips on debugging BiqQuery export jobs - perhaps these stats will be useful for your performance debugging. Let me know if neither of these help. — jkff
@jkff WRT to the first link, we cache the side input on the first call to processElement(). We used to do it from startBundle(), but of course the API changed in on of the last releases, so we moved it to processElement(). — Graham Polley

Davor Bonaci Davor Bonaci · Accepted Answer · 2015-04-23T20:34:53

The evidence seems to indicate that there is an issue with how your pipeline handles side inputs. Specifically, it's quite likely that side inputs may be getting re-read from BigQuery again and again, for every element of the main input. This is completely orthogonal to the changes to the type of virtual machines used by Dataflow workers, described below.

This is closely related to the changes made in the Dataflow SDK for Java, version 0.3.150326. In that release, we changed the side input API to apply per window. Calls to sideInput() now return values only in the specific window corresponding to the window of the main input element, and not the whole side input PCollectionView. Consequently, sideInput() can no longer be called from startBundle and finishBundle of a DoFn because the window is not yet known.

For example, the following code snippet has an issue that would cause re-reading side input for every input element.

@Override
public void processElement(ProcessContext c) throws Exception {
  Iterable<String> uniqueIds = c.sideInput(iterableView);

  for (String item : uniqueIds) {
    [...]
  }

  c.output([...]);
}

This code can be improved by caching the side input to a List member variable of the transform (assuming it fits into memory) during the first call to processElement, and use that cached List instead of the side input in subsequent calls.

This workaround should restore the performance you were seeing before, when side inputs could have been called from startBundle. Long-term, we will work on better caching for side inputs. (If this doesn't help fully resolve the issue, please reach out to us via email and share the relevant code snippets.)

Separately, there was, indeed, an update to the Cloud Dataflow Service around 4/9/15 that changed the default type of virtual machines used by Dataflow workers. Specifically, we reduced the default number of cores per worker because our benchmarks showed it as cost effective for typical jobs. This is not a slowdown in the Dataflow Service of any kind -- it just runs with less resources per worker, by default. Users are still given the options to override both the number of workers as well as the type of the virtual machine used by workers.

Dataflow performance issues

2 Answers