title: DASK: Is there an equivalent of numpy.select for dask?
body: <p>I'm using <strong>Dask</strong> to load an 11m row csv into a dataframe and perform calculations. I've reached a position where I need conditional logic - If this, then that, else other.</p>
  
  <p>If I were to use pandas, for example, I could do the following, where a numpy select statement is used along with an array of conditions and results. This statement takes about 35 seconds to run - not bad, but not great: </p>
  
  <pre><code>df["AndHeathSolRadFact"] = np.select(
      [
      (df['Month'].between(8,12)),
      (df['Month'].between(1,2) &amp; df['CloudCover']&gt;30) #Array of CONDITIONS
      ],  #list of conditions
      [1, 1],     #Array of RESULTS (must match conditions)
      default=0)    #DEFAULT if no match
  </code></pre>
  
  <p>What I am hoping to do is use <strong>dask</strong> to do this, natively, in a dask dataframe, without having to first convert my <strong>dask</strong> dataframe to a pandas dataframe, and then back again. 
  This allows me to:
  - Use multithreading
  - Use a dataframe that is larger than available ram
  - Potentially speed up the result.</p>
  
  <p><strong>Sample CSV</strong></p>
  
  <pre><code>Location,Date,Temperature,RH,WindDir,WindSpeed,DroughtFactor,Curing,CloudCover
  1075,2019-20-09 04:00,6.8,99.3,143.9,5.6,10.0,93.0,1.0 
  1075,2019-20-09 05:00,6.4,100.0,93.6,7.2,10.0,93.0,1.0
  1075,2019-20-09 06:00,6.7,99.3,130.3,6.9,10.0,93.0,1.0
  1075,2019-20-09 07:00,8.6,95.4,68.5,6.3,10.0,93.0,1.0
  1075,2019-20-09 08:00,12.2,76.0,86.4,6.1,10.0,93.0,1.0
  </code></pre>
  
  <p><strong>Full Code for minimum viable sample</strong></p>
  
  <pre><code>import dask.dataframe as dd
  import dask.multiprocessing
  import dask.threaded
  import pandas as pd
  import numpy as np
  
  # Dataframes implement the Pandas API
  import dask.dataframe as dd
  
  from timeit import default_timer as timer
  start = timer()
  ddf = dd.read_csv(r'C:\Users\i5-Desktop\Downloads\Weathergrids.csv')
  
  #Convert back to a Dask dataframe because we want that juicy parallelism
  ddf2 = dd.from_pandas(df,npartitions=4)
  del [df]
  
  print(ddf2.head())
  #print(ddf.tail())
  end = timer()
  print(end - start)
  
  #Clean up remaining dataframes
  del [[ddf2]
  </code></pre>