Merging Tasks

[simon-tarr]

Is there a doAzureParallel 'recommended' way to merge (very) large lists after analysis?

I have a loop of ~500,000 iterations running across a 16-node pool. The Azure processing time within the foreach loop is fast but merging a list that long takes a very long time - much longer than the time taken to carry out the analysis.

I have set a number of arguments which should improve the time taken to return the completed list but they've made very little difference (if any) to the total processing time. These arguments are:

.inorder=F - List order isn't important
.combine = function(...) rbindlist(list(...)) - Apparently rbindlist is faster than the default

I have copied my loop below. It's essentially trying to brute-force the start values for a non-linear model. It is running through all combinations of coefficients (in the object st1.1) until it finds a set of parameters which doesn't result in some sort of convergence error. Those iterations which fail should be removed (.errorhandling=c('remove')) leaving me with a list of potential start values out of the original 500,000.

 mod1.1<-foreach(i = 1:nrow(st1.1), .errorhandling=c('remove'), .inorder=FALSE, .combine = function(...) rbindlist(list(...)), .options.azure = opt, .packages = c("nls2")) %dopar% {
   nls(xx2~b0+b1*exp(b2*xx), data=xx.df, start = list(b0=st1.1[i,1],b1=st1.1[i,2],b2=st1.1[i,3])  ) 
 }

This isn't technically an issue with the package but I have nowhere else to ask such questions in relation to doAzureParallel. Many thanks in advance.

[paselem]

@simon-tarr - this is a good point and a pain-point of merging large runs for a parallel workflow.

Essentially, I think the bottleneck is during the merging phase downloading all of the individual results to a single node for the final aggregation. This is what the process looks like:

Per node, we run a collection of tasks. If your chunk size is greater than 1, we will run multiple iterations on a single R session, which is usually a good way to improve runtimes. Each 'chunk' will produce it's own output which will be uploaded to storage for the merge task to pick up later.

STORAGE    | MERGE TASK ON NODE                      | STORAGE
-----------|-----------------------------------------|------------------------
result_1 --|                                         |
result_2 --|                                         |
result_3 --|-- download individual files and merge --|--> upload merged result
...      --|                                         |
result_n --|                                         |

Having to iterate over 500K files (especially if they are small) is a huge tax to pay on a single node. One recommendation that comes to mind is to do a local reduce on each node (assuming your chunk size is > 1) to reduce the number of files and the download penalty per file. Essentially you will combine several results locally on a single node, reducing the total number of files you upload to storage and the amount of data the merge task needs to process.

The new flow would look like this (assuming a chunk size of 8).

NODE                           | STORAGE  | MERGE TASK ON NODE | STORAGE 
--------|----------------------|----------|-----------------------------------
task_1--|                      |          |                    |
task_2--|   Reduce. Upload     |          |                    |
task_3--|-> single result   -> | result_a |                    |
...   --|   for entire chunk   |          |    download        |
task_8--|                      |          |    individual      |     upload
                                          | -- files           | --> merged
task_9--|                      |          |    and             |     result
task_10-|   Redulce. Upload    |          |    merge           |
task_11-|-> single result   -> | result_b |                    |
...    -|   for entire chunk   |          |                    |
task_16-|                      |          |                    |

This would help reduce the total payload for the merge task to run by ~8x and pre-loads some of the merge work onto every reduce component of each node, distributing that workload across the cluster.

More information on using the reduce is in the man pages for the package:

??setReduce

[simon-tarr]

Hello @paselem thank you for the reply and detailed look as to what's going on.

I've noticed that when all the processing has completed (i.e. the job progress is at 100%) all nodes in the pool are sat at idle, indicating that the processing has completed in the cloud. My local machine's CPU then increases to 10-15% utilisation and will apparently be processing something for hours. After this has completed, my local R session stops processing and the object is fully returned.

Given the node state being idle and the uptick in local CPU usage after the computation (i.e. the foreach status bar reads 100%), I assumed that my computer was carrying out the merging of individual files, not a single node in the pool. Given what you've said above, however, that doesn't seem to be the case?

I've been setting an appropriate chunk size as per the documentation (number of iterations / number of cores). I will try using setReduce this morning to see whether I can improve the overall processing time.

[paselem]

Hi @simon-tarr,
I think I see what you're saying now, which is different to what I originally understood the problem to be. I think there is not too much you can do to optimize the running of doAzureParallel (which is what I previously mentioned). I think this issue is that the piece of data you have to pull down from the cloud is large and takes time to download and load into memory. Do you know the size of the final output from the parallel loop?

If you're not sure, you can find this by looking in the Azure Portal.

Azure Portal -> Azure Storage Account -> Browse Blobs. Type in the name of the job in the 'search by container prefix'. Then simply look for a file called JOB_ID-merge.rds. For example "job20170218071925-merge.rds". It will have the file size next to it. This file will contain the fully merged results of your entire job.

Thanks,
-Pablo

Thanks,
-Pablo

[simon-tarr]

Yeah I think the cloud-side of things in terms of processing is perfect, it's just returning the result which is causing the bother.

Although the original list is large (500k), I've asked the loop to not return any iterations which error. Although really, this doesn't work. If I run the loop locally using a subset of the data (n=10,000 for speed) using %do% I'll get a list returned of length zero (i.e. none of the models converged, therefore each iteration threw an error, therefore nothing was returned). However, running %dopar% and sending to Azure I get returned the convergence error message for each iteration, even when .errorhandling is set to 'remove'. I guess it's possible that the local slowdown I'm seeing is R trying to handle a list of length 500,000!

In response to checking the file sizes: merge.rds is only 37 bytes. The whole job container is less than 100Mb in size.

Thanks
Simon

EDIT - I should be a little clearer. The list returned from Azure using %dopar% is of length zero but R Studio still feeds me data on each of the 10,000 tasks: stdout.txt, stderr.txt and R Output. The R Output file contains the error saying that the loop has failed.

[paselem]

@simon-tarr
We took a look at the code, and it seems like we may not be handling the case where .errorhandling is set to remove and returning results incorrectly. We're looking into it now and will get back to you on it.

[simon-tarr]

Okay great, hopefully it's an easy fix.

[paselem]

Hello @simon-tarr, is this what you're seeing?

What we think might be happening is that downloading all of the error file references (in the Viewer on the right) on your machine is taking a very long time since there are so many of them. We will publish a fix to only download the first 10 error references in the viewer and see if that fixes the issue you're seeing.

[EDIT]
Another potential issue is that you may think that the job is complete because the progress bar is says 100%, but really we're running the merge task, so your job is at 99.999% complete and it is rounding up and reporting being done. Running the merge task on such a large number of iterations is definitely expected to take a little while.

[EDIT]
I tested this locally several times and noticed that the progress bar said 100%, but really when I looked it was 99.93%. I think this is your issue. We will look into a way to make this more explicit. If this is the issue, the previous comments regarding using the .setReduce() function should help improve the efficiency of your runtime significantly.

[simon-tarr]

Hi @paselem with regards to your original reply about downloaded tasks; that's exactly what I'm seeing, yes.
With regards to your second edit, I'm happy to give this a try. However, I don't fully understand the documentation. Where exactly do I tell the loop to use setReduce()? I thought I could use it like this:

 cs <- ceiling(nrow(no.iter) / getDoParWorkers())
 opt <- list(chunkSize = cs, setReduce=cs)

my.loop<-foreach(i = 1:nrow(st1), .errorhandling=c('remove'), .options.azure = opt) %do% {
   myloop()
 }

But I don't think this is correct. Is it an argument in the foreach loop? e.g:

my.loop<-foreach(i = 1:nrow(st1), .errorhandling=c('remove'), .setReduce(cs), .options.azure = opt) %do% {
   myloop()
 }

[paselem]

@simon-tarr,

We have identified one other potential issue in our tools performance and one usability bug (reporting 100% done when it's really not done). These are tracked under #178 and #179.

Regarding using the setReduce() I'm not sure if it will actually help address your issue. I know we've been going back and forth on this (sorry!). We are looking at ways at improving the speed of our merge task and potentially find a way to not pull in data files we probably don't need. These will be bundled into our v0.6.2 which we're targeting to release mid-next week.

W.R.T. to the set reduce, the way to use it is not within the foreach, but just before (similar to registering the back end)

# Run with a local reducer:
doAzureParallel::setReduce(cs)
my.loop<-foreach(i = 1:nrow(st1), .errorhandling=c('remove'), .options.azure = opt) %dopar% {
   myloop()
 }

# Remove the reducer
doAzureParallel::setReduce(NULL)
my.loop<-foreach(i = 1:nrow(st1), .errorhandling=c('remove'), .options.azure = opt) %dopar% {
   myloop()
 }

[simon-tarr]

Hi @paselem thanks for the update. I think I'm going to stop tinkering with getting the current release to work within my workflow and wait until 0.6.2. Many thanks for the communications.

[paselem]

Hi @simon-tarr, we unfortunately are taking a bit longer to release these changes than we'd like. We had to postpone some of the performance improvements until the next release, but will have better monitoring as part of this release (shipping later today as v0.6.2). If possible, please have a look and try it out to see if it helps identify any bottlenecks. Essentially we are trying split up the processing of the parallel tasks vs. the processing of the merge. This should give us a definitive answer around where the bottleneck is in your case.

[simon-tarr]

Hi @paselem I've been trying the new version (v0.6.2 and rBatch 0.5.4) and have seen some improvements in speed when running lists <100,000 long. I'm still encountering some (very) long wait times for anything >~200,000.

Regardless of list length, however, I'm finding that it's taking a very long time for the job to properly start with large lists (which are not large in a file size sense). For example, I can wait hours sometimes after hitting run for the progress bar to appear. I'll be stuck looking at this:

Job Summary: 
Id: job20171212000000

I thought that maybe it's taking some time to upload the list that needs to be processed but the list is only 2.3MB. If I check portal.azure.com all nodes are shown as 'running'. I guess something is happening up in the cloud but I can't see or figure out what that something is.

After the job has run, I will then be sat at 'Merging tasks...' for a very long time. My screen looks something like this at this point (number of failed tasks is as expected):

Job Summary: 
Id: job20171212000000
Job Preparation Status: Package(s) being installed
Waiting for tasks to complete. . .
| Progress: 100.00% (2557/2557) | Running: 0 | Queue: 0 | Completed: 2557 | Failed: 2558 |
Tasks have completed. Merging results.................................................................................................................................................................

At this point, all nodes in my pools are showing as 'idle'.

The good news however is that .errorhandling='remove' is now working as expected (when a job is finally returned).

[paselem]

@simon-tarr, the long time to wait for all tasks to fully render in Batch is expected but there is probably room for improvement. What is happening on your machine is that we're splitting up the work and creating the execution graph in service. Since you have so many tasks it is taking us a long time to build the graph. We will take a look and see if we can improve this performance. Regardless though, your work will start to execute as soon as the first chunk of work is scheduled, so by the time you're done scheduling, you should already see a reasonable amount of progress.

The job sitting at 'Merging tasks...' for a long time confirms that this is indeed the bottleneck in your workflow. I am wondering if we can optimize the run-time by ignoring failed tasks in the merge since you're using the errorhandling='remove' option but I believe it is a non-trivial change. We have already started work on improving the performance though and will spend a few more days finishing it off before we release it to make sure we didn't introduce any regressions.

I'm happy that the errorhandling options are working as expected, but cannot give you any more pointers other than 'wait for our performance fixes' in the next few days to help with you long wait times. We'll loop back around once it's ready.

Thanks for all the feedback!
Pablo

[brnleehng]

Hi @simon-tarr

#194 I've added a PR to speed up merging the tasks by enabling doParallel. If you currently want to test your workload, here's a snippet of code.

devtools::install_github("Azure/doAzureParallel", ref = "fix/merge-task-perf")

Thanks!
Brian

[simon-tarr]

Hi @brnleehng thanks very much! I shall give this a try today and will report back!