How to execute a multi-threaded `merge()` with dask? How to use multiples cores via qsub?
<p>I've just begun using dask, and I'm still fundamentally confused how to do simple pandas tasks with multiple threads, or using a cluster. </p>

<p>Let's take <code>pandas.merge()</code> with <code>dask</code> dataframes. </p>

<pre><code>import dask.dataframe as dd

df1 = dd.read_csv("file1.csv")
df2 = dd.read_csv("file2.csv")

df3 = dd.merge(df1, df2)
</code></pre>

<p>Now, let's say I were to run this on my laptop, with 4 cores. How do I assign 4 threads to this task?</p>

<p>It appears the correct way to do this is:</p>

<pre><code>dask.set_options(get=dask.threaded.get)
df3 = dd.merge(df1, df2).compute()
</code></pre>

<p>And this will use as many threads exist (i.e. as many cores with shared memory on your laptop exist, 4)? How do I set the number of threads?</p>

<p>Let's say I am at a facility with 100 cores. How do I submit this in the same manner as one would submit jobs to the cluster with <code>qsub</code>? (Similar to running tasks on clusters via MPI?)</p>

<pre><code>dask.set_options(get=dask.threaded.get)
df3 = dd.merge(df1, df2).compute
</code></pre>

<h3>Single machine scheduling</h3>

<p>Dask.dataframe will use the threaded scheduler by default with as many threads as you have logical cores in your machine.  </p>

<p>As pointed out in the comments, you can control the number of threads or the Pool implementation with keyword arguments to the <code>.compute()</code> method.</p>

<h3>Distributed machine scheduling</h3>

<p>You can use <a href="http://distributed.readthedocs.io/en/latest/" rel="nofollow">dask.distributed</a> to <a href="http://distributed.readthedocs.io/en/latest/setup.html" rel="nofollow">deploy dask workers across many nodes in a cluster</a>.  One way to do this with <code>qsub</code> is to start a <code>dask-scheduler</code> locally:</p>

<pre><code>$ dask-scheduler
Scheduler started at 192.168.1.100:8786
</code></pre>

<p>And then use <code>qsub</code> to launch many <code>dask-worker</code> processes, pointed at the reported address:</p>

<pre><code>$ qsub dask-worker 192.168.1.100:8786 ... &lt;various options&gt;
</code></pre>

<p>As of yesterday there is an experimental package that can do this on any DRMAA-enabled system (which includes SGE/qsub-like systems): <a href="https://github.com/dask/dask-drmaa" rel="nofollow">https://github.com/dask/dask-drmaa</a></p>

<p>After you have done this you can create a <code>dask.distributed.Client</code> object, which will take over as default scheduler</p>

<pre><code>from dask.distributed import Client
c = Client('192.168.1.100:8786')  # now computations run by default on the cluster
</code></pre>

<h3>Multi-threaded performance</h3>

<p>Note that as of Pandas version 0.19 the GIL still isn't released for <code>pd.merge</code>, so I wouldn't expect a huge speed boost from using multiple threads.  If this is important to you then I recommend putting in a comment here: <a href="https://github.com/pandas-dev/pandas/issues/13745" rel="nofollow">https://github.com/pandas-dev/pandas/issues/13745</a></p>

python multithreading pandas cluster-computing dask