title: Matching groups of data with Dask
body: <h1>Question and Problem statement</h1>
  
  <p>I have data coming from two sources. Each source contains groups identified by <code>ID</code> column, coordinates and attributes. I would like to process this data by first matching these groups, then finding nearest neighbours within these groups, and then studying how the attributes from different sources compare between the neighbors. My learning challenge for myself was how to process this data using parallel processing.</p>
  
  <p>Question is: <em>"Using Dask for parallel processing, what might be the simplest and most straightforward way to process this kind of data?"</em></p>
  
  <h1>Background and my solution thus far</h1>
  
  <p>The data is in CSV files like dummy data below (real files are in the 100 MiB range):</p>
  
  <pre><code>source1.csv:
  ID,X_COORDINATE,Y_COORDINATE,ATTRIB1,PARAM1
  B,-63802.84728184705,-21755.63629150563,3,36.136464492674556
  B,-63254.41147034371,405.6973789009853,1,18.773534321367528
  A,-9536.906537069272,32454.934987740824,0,14.043507555168809
  A,15250.802157581298,-40868.390394552596,0,6.680542212635015
  </code></pre>
  
  <pre><code>source2.csv:
  ID,X_COORDINATE,Y_COORDINATE,ATTRIB1,PARAM1
  B,-6605.150024790153,39733.35763934722,3,5.599467583303852
  B,53264.28797042654,24647.24183964514,0,27.938127686688162
  A,6690.836682554512,34643.0606728128,0,10.02914141165683
  A,15243.16,-40954.928,0,18.130371948545935
  </code></pre>
  
  <p>What I would like to do is to</p>
  
  <ol>
  <li>Load the data into dataframes</li>
  <li>Split them into groups by ID column</li>
  <li>For each group in <code>source1</code> and <code>source2</code>, lets call the sub dataframes in each group <code>source1_sub</code> and <code>source2_sub</code>
  
  <ul>
  <li>construct a kdtree objects <code>k1</code> and <code>k2</code> based on columns X_COORDINATE and Y_COORDINATE</li>
  </ul></li>
  <li>For each pair of objects <code>(k1, k2)</code>
  
  <ul>
  <li>find nearest neighbours for the trees</li>
  <li>construct three dataframes:
  
  <ul>
  <li><code>matches_sub</code>: containing the matched rows in <code>source1_sub</code> and <code>source2_sub</code></li>
  <li><code>source1_sub_only</code>: rows in <code>source1_sub</code> which are not matched</li>
  <li><code>source2_sub_only</code>: rows in <code>source2_sub</code> which are not matched</li>
  </ul></li>
  </ul></li>
  <li>Concatenate all <code>matches_sub</code>, <code>source1_sub_only</code>, and <code>source2_sub_only</code> dataframes into three dataframes: <code>matches</code>, <code>source1_only</code>, <code>source2_only</code></li>
  <li>Analyze these dataframes</li>
  </ol>
  
  <p>This is a problem that should parallelize beautifully, as each pair of groups are independent of other pairs of groups. I decided to use <code>scipy.spatial.cKDTree</code> for the actual coordinate matching, but the difficulty arises from the fact that it operates on indices of the raw numpy arrays, which isn't so straightforwardly compatible with how Dask arrays can be accessed. At least that's my understanding.</p>
  
  <p>My first futile attempts revolved around really awkwardly</p>
  
  <ol>
  <li>Trying to use two Dask dataframes, aligning them and finding matches. This was dreadfully slow and hard to understand.</li>
  <li>Read data with Dask Dataframe and process with using Dask Bag. This was slightly less complex but still not satisfactory.</li>
  </ol>