# Heterogeneous Distributed Big Data Clustering on Sparse Grids
## 5 Implementation
### Distributed Implementation  
For distributed computing, we developed a manager-worker model that was implemented with MPI. To create work that can be assigned to the workers, we split the loops that were used for parallelization (the outermost loops of the computed kernels) once again. We use a static load balancing scheme that distributes the work equally to the workers. Our implementation supports single as well as double precision. Currently, we transfer double precision data even if single precision is used in the computed kernels.  
From the perspective of the manager node, the distributed algorithm consists of four major steps: creating the right-hand side of the density estimation, the density matrix-vector products, an integrated graph-creation-and-prune step and the connected component search. These steps are shown in Figure 5. Note that the matrix-vector multiplication step (Figure 5d) is repeated once per CG iteration.  
When the application is started, the dataset is read by the manager node and sent to all workers, requiring \(m\cdot d\cdot 8\) bytes of communication per worker. Then, the manager node creates the grid and sends it to the workers as well. This requires \(2\cdot N\cdot d\cdot 8\) bytes per worker to be communicated. After these relatively expensive transfers are completed, grid and dataset are held by the workers. Therefore, most remaining communication steps only require small amounts of data to be transferred. We demonstrate in Section 6 that the overhead of these communication steps is indeed very low compared to the computational requirements of the remainder of the algorithm. To read the dataset, we use the standard C++ API. Advanced techniques to improve I/O performance, such as MPI I/O, are currently not used.  
To compute the density right-hand-side operation, every worker computes an index range of the components of \(\mathbf{b}\). As \(\mathbf{b}\) is aggregated on the manager node, \(N\cdot 8\) bytes need to be transferred. During each CG step and after the final CG step, the manager sends \(\mathbf{v}\) (\(\boldsymbol{\alpha}\) after the final iteration) to all workers. Each worker calculates the result of an index range of \(\mathbf{v}^{\prime}\) and communicates the partial result back to the manager node. Therefore, \(N\cdot 8\) bytes per worker are communicated during each iteration and after the final iteration. Collecting the partial results for \(\mathbf{v}^{\prime}\) requires another \(N\cdot 8\) bytes to be transferred per CG step.  
The creation of the \(k\)-nearest-neighbor graph only requires the assignment of index ranges and no further communication. Because the pruning of the \(k\)-nearest-neighbor graph reuses the same  
Figure 5: The distributed clustering algorithm from the perspective of the manager node. The (inexpensive) assignment of index ranges is not shown.  
index ranges that were assigned in the graph creation step, this step only requires the pruned graph to be sent to the manager node. This step requires \(k\cdot m\cdot 8\) bytes to be transferred. Having received the pruned graph, the manager node performs the connected component detection and has thereby computed the clusters.