# Performance of Parallel K-Means Algorithms in Java
## 3 Parallel K-Means in Java
### Supporting K-Means by Streams
load_dataset(); load_centroids(); long start=System.currentTimeMillis(); do{ //Map stage-assign data points to clusters Stream<DataPoint>p_stream=Stream.of(dataset); if(PARALLEL)p_stream=p_stream.parallel(); P_stream.map (p > {  double md=Double.MAX_VALUE; for(int k=0; k<K; ++k) {  double d=p.distance(centroids[k]); if(d<md){md=d; p.setCID(k); } } returnp; }).forEach (p>|); //prepare newCentroids for(int i=0; i<K; ++i) {  new_Centroids[i].reset(); new_Centroids[i].setCID (i); } //Reduce stage-define newCentroids Stream<DataPoint>c_stream=Stream.of(newCentroids); if(PARALLEL)c_stream=c_stream.parallel(); c_stream.map (c > {  for(int i=0; i<N; ++i) {  if(dataset[i].getCID()==c.getCID()) c.add (dataset[i]); }  c.mean(); returnc;}).forEach(c>|); it++; }while(!termination()); long end=System.currentTimeMillis();.../*output operations*/ ```
**Figure 4. A map/reduce schema for K-Means based on Java streams. **  
It is worthy of note that all the globals: the parameters \(N\), \(K\), \(D\), \(T\), \(THR\), \(E\), \(it\), \(MP\), \(PARALLEL\)..., the _dataset_, \(centroids/newCentroids\) data point arrays, the implemented centroid initialization methods (all the ones, either stochastic or deterministic, discussed in [3]), some auxiliary methods for calculating the \(SSD\) cost of a clustering solution or evaluating its quality by, e.g., the _Silhouette_ index [3] and so forth, are introduced as static entities of a \(G\) class, which are statically imported in applicative classes for a direct access to its entities. The Java code in Figure 4 automatically adapts itself to a multi-threaded execution, when the _p_stream_ and _c_stream_ are turned to a parallel version. In this case, behind the scene, Java uses multiple threads of a thread-pool to accomplish in parallel, on selected data segments, the underlying operations required by the Map and Reduce phases. A basic feature of the Java code is its independence from the order of processed data points and the absence of side-effects which can cause interference problems among threadswhen accessing to shared data. During the Map stage, the lambda expression operates on distinct data points and affects their \(CID\) without any race condition. Similarly, multiple threads act on distinct new centroid points during the Reduce phase, and safely accumulate data points of a same cluster. A specific parameter in the \(G\) class is \(INIT\_METHOD\) which takes values in an enumeration with the constants \(RANDOM\), \(MAXIMIN\), \(KMEANSPP\), \(KAUFMAN\), \(ROBIN\), \(DKMEANSPP\), to select a corresponding centroid initialization method [3]. Other initialization methods can be added as well. The ROBIN method, which can be configured to operate stochastically or deterministically, relies concretely on heap-sorting (through a \(PriorityQueue\) collection) the dataset in descending order of the minimal distances from the existing centroids and initially from the reference point. In addition, the nearest \(MP\)-neighborhood of a data point \(x_{i}\), considered in sorted order, is determined by moving around \(x_{i}\), and registering in a sorted data structure the distances of nearest points to \(x_{i}\). Movement is stopped as soon as a point is encountered whose distance from \(x_{i}\) is greater than that of any already met nearest point, and the cardinality of previous visited nearest points is found to be \(\geq MP\). The \(G\) class is extended by a further parameter \(P\) which specifies the degree of parallelism, that is the number of the underlying cores (threads) exploitable in the execution experiments.