We have implemented the serial version of the algorithm in the KnnClassifier class. Internally, this class stores the train dataset and the number k (the number of examples that we will use to determine the tag of an instance):

public class KnnClassifier {

  private List <? extends Sample> dataSet;
  private int k;

  public KnnClassifier(List <? extends Sample> dataSet, int k) {
    this.dataSet=dataSet;
    this.k=k;
  }

The KnnClassifier class only implements a method named classify that receives an Sample object with the instance we want to classify, and it returns a string with the tag assigned to that instance:

  public String classify (Sample example) {

This method has three main parts—first, we calculate the distances between the input example and all the examples of the train dataset:

    Distance[] distances=new Distance[dataSet.size()];

    int index=0;

    for (Sample localExample : dataSet) {
      distances[index]=new Distance();
      distances[index].setIndex(index);
      distances[index].setDistance (EuclideanDistanceCalculator.calculate(localExample, example));
      index++;
    }

Then, we sort the examples from lower to higher distance, using the Arrays.sort() method:

    Arrays.sort(distances);

Finally, we count the tag with most instances in the k-nearest examples:

    Map<String, Integer> results = new HashMap<>();
    for (int i = 0; i < k; i++) {
      Sample localExample = dataSet.get(distances[i].getIndex());
      String tag = localExample.getTag();
      results.merge(tag, 1, (a, b) -> a+b);
    }
    return Collections.max(results.entrySet(), Map.Entry.comparingByValue()).getKey();
  }

To calculate the distance between two examples, we can use the Euclidean distance implemented in an auxiliary class. This is the code of that class:

public class EuclideanDistanceCalculator {
  public static double calculate (Sample example1, Sample example2) {
    double ret=0.0d;

    double[] data1=example1.getExample();
    double[] data2=example2.getExample();

    if (data1.length!=data2.length) {
      throw new IllegalArgumentException ("Vector doesn't have the same length");
    }

    for (int i=0; i<data1.length; i++) {
      ret+=Math.pow(data1[i]-data2[i], 2);
    }
    return Math.sqrt(ret);
  }

}

We have also used the Distance class to store the distance between the Sample input and an instance of the train dataset. It only has two attributes: the index of the example of the train dataset and the distance to the input example. In addition, it implements the Comparable interface to use the Arrays.sort() method. Finally, the Sample class stores an instance. It only has an array of doubles and a string with the tag of that instance.

If you analyze the serial version of the k-nearest neighbors algorithm, you can find the following two points where you can parallelize the algorithm:

In the first concurrent version of the algorithm, we are going to create a task per distance between examples that we're going to calculate. We are also going to make it possible to produce a concurrent sort of array of distances. We have implemented this version of the algorithm in a class named KnnClassifierParrallelIndividual. It stores the train dataset, the k parameter, the ThreadPoolExecutor object to execute the parallel tasks, an attribute to store the number of worker-threads we want to have in the executor, and an attribute to store if we want to make a parallel sort.

We are going to create an executor with a fixed number of threads so that we can control the resources of the system that this executor is going to use. This number will be the number of processors available in the system we obtain with the availableProcessors() method of the Runtime class multiplied by the value of a parameter of the constructor named factor. Its value will be the number of threads you will have from the processor. We will always use the value 1, but you can test with other values and compare the results. This is the constructor of the classification:

public class KnnClassifierParallelIndividual {

  private List<? extends Sample> dataSet;
  private int k;
  private ThreadPoolExecutor executor;
  private int numThreads;
  private boolean parallelSort;

  public KnnClassifierParallelIndividual(List<? extends Sample> dataSet, int k, int factor, boolean parallelSort) {
    this.dataSet=dataSet;
    this.k=k;
    numThreads=factor* (Runtime.getRuntime().availableProcessors());
    executor=(ThreadPoolExecutor) Executors.newFixedThreadPool(numThreads);
    this.parallelSort=parallelSort;
  }

To create the executor, we have used the Executors utility class and its newFixedThreadPool() method. This method receives the number of worker-threads you want to have in the executor. The executor will never have more worker-threads than the number you specified in the constructor. This method returns an ExecutorService object, but we cast it to a ThreadPoolExecutor object to have access to methods provided but the class and not included in the interface.

This class also implements the classify() method that receives an example and returns a string.

First, we create a task for every distance we need to calculate and send them to the executor. Then, the main thread has to wait for the end of the execution of those tasks. To control that finalization, we have used a synchronization mechanism provided by the Java concurrency API: the CountDownLatch class. This class allows a thread to wait until other threads have arrived at a determined point of their code. It's initialized with the number of threads you want to wait for. It implements two methods:

In this case, we initialize the CountDownLatch class with the number of tasks we are going to execute in the executor. The main thread calls the await() method and calls the getDown() method for every task, when it finishes its calculation:

  public String classify (Sample example) throws Exception {

    Distance[] distances=new Distance[dataSet.size()];
    CountDownLatch endController=new CountDownLatch(dataSet.size());

    int index=0;
    for (Sample localExample : dataSet) {
      IndividualDistanceTask task=new IndividualDistanceTask(distances, index, localExample, example, endController);
      executor.execute(task);
      index++;
    }
    endController.await();

Then, depending on the value of the parallelSort attribute, we call the Arrays.sort() or Arrays.parallelSort() methods.

    if (parallelSort) {
      Arrays.parallelSort(distances);
    } else {
      Arrays.sort(distances);
    }

Finally, we calculate the tag assigned to the input examples. This code is the same as in the serial version.

The KnnClassifierParallelIndividual class also includes a method to shutdown the executor calling its shutdown() method. It you don't call this method, your application will never end because threads created by the executor are still alive and waiting for the new tasks to do. Previously submitted tasks are executed, and newly submitted tasks are rejected. The method doesn't wait for the finalization of the executor, it returns immediately:

  public void destroy() {
    executor.shutdown();
  }

A critical part of this example is the IndividualDistanceTask class. This is the class that calculates the distance between the input example and an example of the train dataset as a concurrent task. It stores the full array of distances (we are going to establish the value of one of its positions only), the index of the example of the train dataset, both examples, and the CountDownLatch object used to control the end of the tasks. It implements the Runnable interface, so it can be executed in the executor. This is the constructor of the class:

public class IndividualDistanceTask implements Runnable {

  private Distance[] distances;
  private int index;
  private Sample localExample;
  private Sample example;
  private CountDownLatch endController;

  public IndividualDistanceTask(Distance[] distances, int index, Sample localExample,
      Sample example, CountDownLatch endController) {
    this.distances=distances;
    this.index=index;
    this.localExample=localExample;
    this.example=example;
    this.endController=endController;
  }

The run() method calculates the distance between the two examples using the EuclideanDistanceCalculator class explained before and stores the result in the corresponding position of the distances:

  public void run() {
    distances[index] = new Distance();
    distances[index].setIndex(index);
    distances[index].setDistance (EuclideanDistanceCalculator.calculate(localExample, example));
    endController.countDown();
  }

The concurrent solution presented in the previous section may have a problem. You are executing too many tasks. If you stop to think, in this case, we have more than 29,000 train examples, so you're going to launch 29,000 tasks per example you want to classify. On the other hand, we have created an executor with a maximum of numThreads worker-threads, so another option is to launch only numThreads tasks and split the train dataset in numThreads groups. We executed the examples with a quad-core processor, so each task will calculate the distances between the input example and approximately 7,000 train examples.

We have implemented this solution in the KnnClassifierParallelGroup class. It's very similar to the KnnClassifierParallelIndividual class with two main differences. First, the first part of the classify() method. Now, we will only have numThreads tasks, and we have to split the train dataset in numThreads subsets:

  public String classify(Sample example) throws Exception {

    Distance distances[] = new Distance[dataSet.size()];
    CountDownLatch endController = new CountDownLatch(numThreads);

    int length = dataSet.size() / numThreads;
    int startIndex = 0, endIndex = length;

    for (int i = 0; i < numThreads; i++) {
      GroupDistanceTask task = new GroupDistanceTask(distances, startIndex, endIndex, dataSet, example, endController);
      startIndex = endIndex;
      if (i < numThreads - 2) {
       endIndex = endIndex + length;
      } else {
       endIndex = dataSet.size();
      }
      executor.execute(task);

    }
    endController.await();

We calculate the number of samples per task in the length variable. Then, we assign to each thread the start and end indexes of the samples they have to process. For all the threads except the last one, we add the length value to the start index to calculate the end index. For the last one, the last index is the size of the dataset.

Second, this class uses GroupDistanceTask instead of IndividualDistanceTask. The main difference between those classes is that the first one processes a subset of the train dataset, so it stores the full train dataset and the first and last positions of the dataset it has to process:

public class GroupDistanceTask implements Runnable {
  private Distance[] distances;
  private int startIndex, endIndex;
  private Sample example;
  private List<? extends Sample> dataSet;
  private CountDownLatch endController;

  public GroupDistanceTask(Distance[] distances, int startIndex, int endIndex, List<? extends Sample> dataSet, Sample example, CountDownLatch endController) {
    this.distances = distances;
    this.startIndex = startIndex;
    this.endIndex = endIndex;
    this.example = example;
    this.dataSet = dataSet;
    this.endController = endController;
  }

The run() method processes a set of examples instead of only one example:

  public void run() {
    for (int index = startIndex; index < endIndex; index++) {
      Sample localExample=dataSet.get(index);
      distances[index] = new Distance();
      distances[index].setIndex(index);
        distances[index].setDistance(EuclideanDistanceCalculator
            .calculate(localExample, example));
    }
    endController.countDown();
  }

Let's compare the different versions of the k-nearest neighbors algorithms we have implemented. We have the following five different versions:

To test the algorithm, we have used 2,059 test instances, which we take from the Bank Marketing dataset. We have classified all those examples using the five versions of the algorithm using the values of k as 10, 30, and 50, and measured their execution time. We have used the JMH framework (http://openjdk.java.net/projects/code-tools/jmh/) which allows you to implement microbenchmarks in Java. Using a framework for benchmarking is a better solution that simply measures time using the currentTimeMillis() or nanoTime() methods. These are the results:

We can draw the following conclusions:

So, the best version of the algorithm is the coarse-grained solution using parallel sorting. If we compare it with the serial version calculating the speedup:

This example shows how a good election of a concurrent solution can give us a great improvement, and a bad election can give us a bad performance.