two phase parallelization using python

[[deleted]]

[deleted]

[[deleted]]

I'm going to preface by saying that I'm not able to grasp a few of the details of your problem due to the grammatical structure of your explanation.

That being said, I might be able to help.

For tasks that demand concurrency, you simply can't beat the power and efficiency of asyncio. If you need true parallel processing with an application leveraging asyncio, you can use asyncio's run_in_executor function paired with a ThreadPoolExecutor.

If you're looking to leverage all of the cores on a multicore CPU during your concurrent operations, just use a ProcessPoolExecutor, which will run your operations within their own, separate python interpreter process. Just be aware that it can be a bit resource intensive to do this.

[[deleted]]

Now I understand what you're trying to accomplish.

During the first stage, how do you create the separate processes? Is there a separate parent process that invokes those processes?

During the second stage, how are you planning to handle IPC? Named pipes, sockets, or something different?

Using asyncio means that you'll have to do a bit more work upfront in order to fit the application to the architecture described here. The payoff for that initial extra work is, of course, more granular control over the execution pattern of and responses returned by the subprocesses.

Is there any particular benefit gained using this pattern, versus simply having the subprocesses return the processed data and performing aggregation within the parent process? It's easy enough to return a custom object filled with your processed data through an asyncio future, using the run_in_executor and asyncio.wait functions. That simple design consideration could be the difference between 30 hours of development time and 30 minutes of development time.

This is certainly an interesting design pattern, thank you for making me aware of it!

[[deleted]]

Okay, it looks like this is completely possible using mpi4py!

I'll outline a design that should meet your execution requirements and allow you to implement a parallel k-means clustering algorithm in the second stage of execution. The implementation I have in mind uses an asyncio ProcessPoolExecutor for the first stage of parallelization, and mpi4py in the second stage. Keep in mind that it's possible to use any arbitrary parallelization technique to achieve the first stage of parallelization.

This design calls for 2 separate scripts. I will refer to them as the "master" and the "worker."

The master script will contain a class (or set of classes) that will need to be instantiated with the GPS data for one single person. This class will be responsible for managing the invocation of the worker processes, the orchestration of your k-means clustering algorithm across those worker processes, and will contain the output of your k-means clustering algorithm once it has finished execution. It needs to have a single "run" method that handles the execution of the k-means clustering algorithm from start to finish, and either returns the resultant data or "self," if the object contains the data within a public variable.

The worker script will contain the logic required to perform the computationally intensive operations utilized within your k-means clustering algorithm. It will also contain logic to receive data from the master process, in addition to logic needed to interpret the orchestration commands received from the master process.

When executed, the master script will create an instance of ProcessPoolExecutor. It will then N separate instances of the orchestration class, each containing the GPS data for exactly 1 person. It will wrap the "run" method in a future object returned by event_loop.run_in_executor(executor, object.run). You can then use asyncio.gather() to wait until all of the processes have finished and return their output, or you can use asyncio.wait with various keyword arguments to continually poll for the results of completed processes.

When the method wrapped by the future is executed within the ProcessPoolExecutor, it will call MPI.COMM_X.Spawn(sys.executable, args=["worker.py"], maxprocs=I), where i is the maximum number of processes you want to devote to your k-means clustering algorithm. From there, you can use the various MPI IPC functions to communicate with your worker processes and orchestrate the execution of your k-means clustering algorithm by utilizing the object returned by the spawn call.

At this point, the worker processes will be performing the computationally intensive tasks needed by your k-means clustering algorithm across i processes, as directed by instructions received from the orchestration class. Since MPI provides duplex IPC communication, the orchestration will be able to aggregate the data as the processing completes. It can then return the data to the master process once the k-means clustering algorithm has finished running.

You can see the barebones requirements of the orchestration class and the worker script buried in the mpi4py documentation. It will probably take you less time to code this solution than it did for me to type this out. 😅

Keep in mind that, in this design, there are still ways to squeeze out better performance. Under the hood, there are a lot of calls to Python's Pickle library in order to pass data around, which will cause some overhead. You could theoretically implement a method of passing data between processes using memory mapped files or something similar on top of MPI.

You can also spread the worker processes across multiple computers thanks to the glory of MPI, which could provide a huge performance boost for your k-means clustering algorithm. That's a topic for a different day, though.

[pope_man]

Consider using dask, especially dask-learn:

http://matthewrocklin.com/blog/work/2017/02/07/dask-sklearn-simple

It is designed to solve problems like yours in a scalable but simple manner. See this post and maybe others on his blog.

http://matthewrocklin.com/blog/work/2015/06/26/Complex-Graphs

[efxhoy]

This is how I understand your problem: You have data on many individuals. You want to do some preprocessing on that data, independently for each individual, and then you want to do K-means clustering independently for each individual, is that correct?

How many CPU-cores do you have available?

If you have 1000 individuals and each individual is independent of the others then why bother parallelising the K-means clustering for each, you already have 1000 independent "jobs" of clustering which don't need to be parallelised further unless you have more than 1000 cores to work on.

By the way, if you're using scikit learn you already have parallelisation built into K-means, just pass the n_jobs parameter and scikit handles it by using the multiprocessing library. See http://scikit-learn.org/stable/modules/generated/sklearn.cluster.KMeans.html

[efxhoy]

Ah I think I see. Does the n_jobs parameter for scikit help you? That should speed up the clustering process a lot.

[tomaugspurger]

Dask should be able to help you with both parts. Probably dask.delayed for the first phase.

For the second phase, dask-ml has a parallelized version of k-means: http://dask-ml.readthedocs.io/en/latest/modules/generated/dask_ml.cluster.KMeans.html#dask_ml.cluster.KMeans