ensemble processing idea for dask distributed

[pavlis]

While working on the upcoming Earthscope short course on MsPASS I was reading through the new documentation on dask distributed. It made me realize there as a potentially very useful model for processing large ensembles that could both improve throughput and reduce the likelihood of memory bloat killing a job.

In a nutshell, the idea is to use the newer approach for concurrent computing using their "Futures" object described here. How to use a "Futures" in MsPASS is the question that I had a bit of a epiphany on. The idea simply is that if you need to process the members of a large ensemble, submit the algorithm as a "Futures" that returns changes the ensemble (or reduces it) and runs the algorithm asynchronously. In that model, I think you could run map operations on atomic members and call compute at the end of the function submitted as a Futures. The scheduler would treat that set of tasks like any task in our normal map-reduce model, but the atomic tasks could be handled by the scheduler and reduce memory bloat.

I think the basic code would be to define an ensemble function like this:

def example_atomic(ensemble,i):
   ensemble.member[i]=some_atomic_function(ensemble.member[i])
   return i
def example_futures(ensemble):
   ilist = list(range(len(ensemble.member)):
   mybag = bag.from_sequence(ilist)
   mybag = mybag.map(example_atomic)
   mybag.compute() 

Now that example may not be exactly right. I'm not sure the assignments will work correctly in the example_atomic function. The real point is using that as a Futures that would be typically done something like this:

futures_list = []
srcid = db.source.distinct('_id')
for id in srcid:
   cursor = db.wf_TimeSeries.find({'source_id' : id})
   ensemble = db.read_data(cursor)
   this_future = client.submit(example_futures,ensemble)
   futures_list.append(this_future)
# have no idea whether or not this trick will work
fmap = dask.bag.from_sequence(futures_list)
fmap=fmap.map(result())
fmap=fmap.map(db.save_data)   # default now returns a small memory list 
fmap.compute()

In that algorithm a big unknown is if dask would handle the mix of futures with a bag in scheduling. That is, however, only one way to exploit this idea. The big point is the generic idea of using submit to send a large ensemble to the cluster to be processed assynchronously.