dataesr/scientific-paragraphs-categorization huggingface dataset:
The breakthrough of this paper is that the solutions are experimentally compared in the same setting: same hardware and same dataset. Moreover, we present in this paper experimental settings and configurations that were not studied in the original papers. Overall, our contributions are: • The decomposition of a distributed MapReduce kNN computation in different basic steps, introduced in Section 3. • A theoretical comparison of existing techniques in Section 4, focusing on load balancing, accuracy and complexity aspects. • An implementation of 5 published algorithms and an extensive set of experiments using both low and high dimension datasets (Section 5). • An analysis which outlines the influence of various parameters on the performance of each algorithm. The paper is concluded with a summary that indicates the typical dataset-solution coupling and provides precise guidelines to choose the best algorithm depending on the context.
MapReduce MapReduce  [13]  is a parallel programming model that aims at efficiently processing large datasets. This programming model is based on three concepts: (i) representing data as key-value pairs, (ii) defining a map function, and (iii) defining a reduce function. The map function takes keyvalue pairs as an input, and produces zero or more keyvalue pairs. Outputs with the same key are then gathered together (shuffled) so that key-{list of values} pairs are given to reducers. The reduce function processes all the values associated with a given key. The most famous implementation of this model is the Hadoop framework 2  which provides a distributed platform for executing MapReduce jobs.
Existing kNN Computing Systems The basic solution to compute kNN adopts a nested loop approach, which calculates the distance between every object r i in R and s j in S and sorts the results to find the k smallest. This approach is computational intensive, making it unpractical for large or intricate datasets. Two strategies have been proposed to work out this issue. The first one consists in reducing the number of distances to compute, by avoiding scanning the whole dataset. This strategy focuses on indexing the data through efficient data structures. For example, a one-dimension index structure, the B + -Tree, is used in  [17]  to index distances;  [18]  adopts a multipage overlapping index structure R-Tree;  [10]  proposes to use a balanced and dynamic M-Tree to organize the dataset;  [19]  introduces a sphere-tree with a sphere-shaped minimum bound to reduce the number of areas to be searched;  [20]  presents a multidimensional quad-tree in order to be able to handle large amount of data;  [12]  develops a kd-tree which is a clipping partition method to separate the search space; and  [21]  introduces a loose coupling and shared nothing distributed Inverted Grid Index structure for processing kNN query on MapReduce.
However, reducing the searched dataset might not be sufficient. For data in large dimension space, computing the distance might be very costly. That is why a second strategy focuses on projecting the highdimension dataset onto a low-dimension one, while maintaining the locality relationship between data. Representative efforts refer to LSH (Locality-Sensitive Hashing)  [22]  and Space Filling Curve  [23]. But with the increasing amount of data, these methods still can not handle kNN computation on a single machine efficiently. Experiments in  [24]  suggest using GPUs to significantly improve the performance of distance computation, but this is still not applicable for large datasets that cannot reasonably be processed on a single machine. More recent papers have focused on providing efficient distributed implementations.
Data Partitioning and Selection MapReduce is a shared-nothing platform, so in order to process data on MapReduce, we need to divide the dataset into independent pieces, called partitions. When computing a kNN, we need to divide R and S respectively. As in any MapReduce computation, the data partition strategy will strongly impact CPU, network communication and disk usages, which in turn will impact the overall processing time  [31]. Besides, a good partition strategy could help to reduce the number of data replications, thereby reducing the number of distances needed to be calculated and sorted. However, not all the algorithms apply a special data partition strategy. For example, H-BNLJ simply divides R into rows and S into lines, making each subset of R meeting with every subset of S. This ensures the distance between each object r i in R and each object s j in S will be calculated.
This way of dividing datasets causes a lot of data replications. For example, in H-BNLJ, each piece of data is duplicated n times (n is the number of subsets of R and S), resulting in a total of n 2 tasks to calculate pairwise distances. This method wastes a lot of hardware resources, and ultimately leads a low efficiency. The key to improve the performance is to preserve spatial locality of objects when decomposing data for tasks  [32]. This means making a coarse clustering in order to produce a reduced set of neighbors that are candidates for the final result. Intuitively, the goal is to have a partitioning of data such that an element in a partition of R will have its nearest neighbors in only one partition of S. Two partitioning strategies that enable to separate the datasets into independent partitions, while preserving locality information, have been proposed.
Distance Based Partitioning Strategy The distance based partitioning strategy we study in this paper is based on Voronoi diagram, a method to divide the space into disjoint cells. We can find other distancebased partitioning methods in the litterature, such as in  [21] , but we chose Voronoi diagram to represent distance-based partitioning method because it it can apply to data in any dimension. The main property of Voronoi diagram is that every point in a cell is closer to the pivot of this cell than to any other pivot. More formally, the definition of a Voronoi cell is as follow: Definition 4. Given a set of disjoint pivots: P = {p 1 , p 2 ,. , p i ,. , p n }, the Voronoi Cell of p i (0 < i ≤ n) is: ∀ i = j, V C (p i ) = {p d (p, p i ) ≤ d (p, p j )}. Paper  [28]  gives a method to partition datasets R and S using Voronoi diagram.
The partitioning principles are illustrated in Figure  1. After having identified the pivots p i in R (c. f. Section 3. 1), the distances between elements of each dataset and the pivots are computed. The elements are then put in the cell of the closest pivot, giving a partitioning of R (resp. S) into P R i (resp. P S i ). For each cell, the upper bound U (P R i ) (resp. the lower bound L(P R i )) is computed as a sphere determined by the furthest (resp. nearest) point in P R i from the pivot p i. The boundaries and other statistics are used to find candidate data from S in the neighboring cells. These data are then replicated in cell P S i. For example, in Figure  1 , the element s of P S j falls inside U (P R i ) and is thus copied to S i as a potential candidate for the kNN of r. The main issue with this method is that it requires computing the distance from all elements to the pivots.
Size Based Partitioning Strategy Another type of partitioning strategy aims at dividing data into equal size partitions. Paper  [29]  proposes a partitioning strategy based on z-value described in the previous section. In order to have a similar number of elements in all n partitions, the authors first sample the dataset and compute the n quantiles. These quantiles are an unbiased estimation of the bounds for each partition. Figure  2  shows an example for this method. In this example data are only shifted once. Then, data are projected using z-value method, and the interpolating "Z" in the figure indicates the neighborhood after projection. Data is projected into a one dimension space, represented by Z R i and Z S i in the figure. Z R i is divided into partitions using the sampling estimation explained above.
One MapReduce Job Without preprocessing and partitioning strategies. The naive solution (H-BkNNJ) only uses one MapReduce job to calculate and sort the distances, and only the Map Phase is done in parallel. The Map tasks will cut datasets into splits, and label each split with its original dataset (R or S). The Reduce task then takes one object r i and one object s j to form a key-value pair < r i , s j >, and calculate the distance between them, then for each key r i sort the distances with every object in S, leading the number of distances need to be sorted to |S|. Since only the Map phase is in parallel, and only one Reduce task is used for calculating and sorting, when the datasets becomes large, this method will quickly exceed the processing capacity of the computer. Therefore, it is only suitable for small datasets.
Then each Reduce task will sort list(s id , d(r, s)) in ascending order of d(r, s) for each r i , and finally, give the top k results. Moreover, in order to avoid the scan of the whole dataset of each block, some index structures like R-Tree  [29]  or Hilbert R-Tree  [33]  can be used to index the local S blocks. With preprocessing and partitioning strategies. In H-zkNNJ  [29]  the authors propose to define the bounds of the partitions of R and then to determine from this the corresponding S i in a preprocessing job. So here, the preprocessing and partitioning steps are completely integrated in MapReduce. Then, a second MapReduce job takes the partitions R i and S i previously determined, and computes for all r i the candidate neighbor set, which represents the points that could be in the final kNN  3.
Load Balance In a MapReduce job, the Map tasks or the Reduce tasks will be processed in parallel, so the overall computation time of each phase depends on the completion time of the longest task. Therefore, in order to obtain the best performance, it is important that each task performs substantially the same amount of computation. When considering load balancing in this section, we mainly want to have the same time complexity in each task. Ideally, we want to calculate roughly the same number of distance between r i and s j in each task. For H-BkNNJ, there is no load balancing problem. Because in this basic method, only the Map Phase is treated in parallel. In Hadoop each task will process 64M data by default. H-BNLJ cuts both the dataset R and the dataset S into p equal-size pieces, then those pieces are combined pairwise to form a partition of < R i , S j >.
Accuracy Usually, the lack of accuracy is the direct consequence of techniques to reduce the dimensionality with techniques such as z-values and LSH. In  [29]  (H-zkNNJ), the authors show that when the dimension of the data increases, the quality of the results tends to decrease. This can be counterbalanced by increasing the number of random shifts applied to the data, thereby increasing the size of the resulting dataset. Their experiments show that three shifts of the initial dataset (in dimension 15) are sufficient to achieve a good approximation (less than 10% of errors measured), while controlling the computation time. Furthermore, paper  [34]  shows a detailed theoretical analyses showing that, for any fixed dimension, by using only O(1) random shifts of data, the z-value method returns a constant factor approximation in terms of the radius of the k nearest neighbor ball.
(3) The number of final candidates for each object r i : We have seen that advanced algorithms use preprocessing and partitioning techniques to reduce this number as much as possible. The goal is to reduce the amount of data transmitted and the computational cost. Together these three points impact two main overheads that affect the performance: • Communication overhead, which can be considered as the amount of data transmitted over the network during the shuffle phases. • Computation overhead, which is mainly composed of two parts: 1). computing the distances, 2). finding the k smallest distances. It is also impacted by the complexity (dimension) of the data. Suppose the dataset is d dimensional, the overhead for computing the distance is roughly the same for every r i and s j for each method.
The difference comes from the number of distances to sort for each element r i to get the top k nearest neighbors. Suppose that the dataset R is divided into n splits. Here n represents the number of partitions of R and S for H-BNLJ and H-zkNNJ, the number of cells after using the grouping strategy for PGBJ and the number of buckets for RankReduce. Assuming there is a good load balance for each method, the number of elements in one split R i can be considered as |R| n. Finding the k closest neighbors efficiently for a given r i can be done using a priority queue, which less costly than sorting all candidates. Since all these algorithms uses different strategies, their steps cannot be directly compared. Nonetheless, to provide a theoretical insight, we will now compare their complexity for the last phase, which is common to all of them.
The basic method H-BkNNJ only uses one MapReduce job, and requires only one Reduce task to compute and sort the distances. The communication overhead is O(|R| + |S|). The number of final candidates for one r i is |S|. The complexity of finding the k smallest distances for r i is O(|S| • log (k)). Hence, the total cost for one task is O(|R|•|S|•log (k)). Since R and S are usually large datasets, this method quickly becomes impracticable. To overcome this limitation, H-BNLJ  [29]  uses two MapReduce jobs, with n 2 tasks to compute the distances. Using a second job significantly reduces the number of final candidates to nk. The total communication overhead is O(n |R| + n |S| + kn |R|). The complexity of finding the k elements for each r i is reduced to (n • k) • log (k). Since each task has |R| n elements, the total sort overhead for one task is O(|R| • k • log(k)).
PGBJ  [28]  performs a preprocessing phase followed by two MapReduce jobs. This method also only uses n Map tasks to compute the distances and the number of final candidates falls to |S i |. Since this method uses a distance based partitioning method, the size of |S i | varies, depending on the number of cells required to perform the computation and the number of replications (|RepS c |, see Section 4. 1) required by each cell. As such, the computational complexity cannot be expressed easily. Overall, finding the k elements is reduced to O(|S i | • log |k|) for each r i , and O( |R| n • |S i | • log |S i |) in total per task. The communication overhead is O(|R| + |S| + |RepS c | • n). In the original paper the authors gives a formula to compute |RepS c | • n, which is the total number of replications for the whole dataset S.
In RankReduce  [27] , the initial dataset is projected by L hash families into buckets. After finding the local candidates in the second job, the third job combines the local results to find the global k nearest neighbor. For each r i , the number of final candidates is L • k. Finding the k elements takes O(L • k • log(k)) per r i , and O(|R i | • L • k • log(k)) per task. The total communication cost becomes O(|R| + |S| + k • |R|). H-zkNNJ  [29]  also begins by a preprocessing phase and uses in total three MapReduce jobs in exchange for requiring only n Map tasks. For a given r i , these tasks process elements from the candidate neighbor set C (r i ). By construction, C (r i ) only contains α • k neighbors, where α is the number of shifts of the original dataset. The complexity is now reduced to O((α • k) • log (k)) for one r i , and O( |R| n • (α • k) • log (k)) in total per task.
First, the least efficient should be H-BkNNJ, followed by H-BNLJ. PGBJ, RankReduce and H-zkNNJ are theoretically the most efficient. Among them, PGBJ has the largest number of final candidates. For RankReduce and H-zkNNJ , the number of final candidates is of the same order of magnitude. The main difference lies in the communication complexity, more precisely in 1 ε 2 compared to |R|. As the dataset size increases, we will have eventually |R| 1 ε 2. Hence, H-zkNNJ seems to be the theoretically most efficient for large query sets.
EVALUATION In order to compare theoretical performance and performance in practice, we performed an extensive experimental evaluation of the algorithms described in the previous sections. The experiments were run on two clusters of Grid'5000 5 , one with Opteron 2218 processors and 8GB of memory, the other with Xeon E5520 processors and 32GB of memory, using Hadoop 1. 3, 1Gb/s Ethernet and SATA hard drives. We follow the default configuration of Hadoop: (1) the number of replications for each split of data is set to 3; (2) the number of slot of each node is 1, so only one map/reduce task is processed on the node at one time. We evaluate the five approaches presented before. For H-zkNNJ and H-BNLJ, we took the source code provided by the authors as a starting point 6 and added some modifications to reduce the size of intermediate files.
The others were implemented from scratch using the description provided in their respective papers. When implementing RankReduce, we added a reduce phase to the first MapReduce job to compute some statistics information for each bucket. These information is used for achieving good load balance. Moreover, in the original version, the authors only use one hash function. To improve the precision, we choose to use multiple families and hash functions depending on the dataset. Finally, our version of RankReduce uses three MapReduce jobs instead of two. Most of the experiments were ran using two different datasets: • OpenStreetMap: we call it the Geographic -or Geodataset. The Geo dataset involves geographic XML data in two dimensions 7. This is a real dataset containing the location and description of objects.
The data is organized by region. We extract 256 * 10 5 records from the region of France. • Catech 101: we call it the Speeded Up Robust Featuresor SURF -dataset. It is a public set of images 8 , which contains 101 categories of pictures of objects, and 40 to 800 images per category. SURF  [37]  is a detector and a descriptor for points of interest in images, which produces image data in 128 dimensions. We extract 32 images per category, each image has between 1000 and 2000 descriptors. In order to learn the impact of dimension and dataset, we use 5 additional datasets: El Nino: in 9 dimensions; HIGGS: in 28 dimensions; TWITTER: in 77 dimensions; BlogFeedBack: in 281 dimensions; and Axial Axis: in 386 dimensions. These data sets are all downloaded from UCI Machine Learning Repository 9. We use a self-join for all our experiments, which means we use two datasets of equal sizes for R and S (|R| = |S|).
We also vary the number of records of each dataset from 6. http://ww2. cs. fsu. edu/ ∼ czhang/knnjedbt/ 7. Taken from: http://www. geofabrik. de/data/download. html 8. Taken from: www. vision. caltech. edu/Image Datasets/Caltech101 9. Taken from: https://archive. ics. uci. edu/ml/ 0. 125 * 10 5 to 256 * 10 5. For all experiments, we have set k = 20 unless specified otherwise. We study the methods from the following aspects: • The impact of data size • The impact of k • The impact of dimension and dataset And we record the following information: the processing time, the disk space required, the recall and precision, and the communication overhead. To assess the quality of the approximate algorithms, we compute two commonly used metrics and use the results of the exact algorithm PGBJ as a reference.
We thus have space = Size f inal +Size intermediate Sizecorrect. We start by evaluating the most efficient number of machines to use (hereafter called nodes) in terms of resources and computing time. For that, we measure the computing time of all algorithm for three different data input size of the geographic dataset. The result can be seen on Figure  5. As expected, the computing time is strongly related to the number of nodes. Adding more nodes increases parallelism, reducing the overall computing time. There is however a significant slow down after using more than 15 machines. Based on those results, and considering the fact that we later use larger datasets, we conducted all subsequent experiments using at most 20 nodes.
Impact of input data size Our first set of experiments measures the impact of the data size on execution time, disk space and recall. Figure  6a  shows the global computing time of all algorithms, varying the number of records from 0. 125 * 10 5 to 256 * 10 5. The global computing time increases more or less exponentially for all algorithms, but only H-zkNNJ and RankReduce can process medium to large datasets. For small datasets, PGBJ can compute an exact solution as fast as the other algorithms. Figure  6b  shows the space requirement of each algorithm as a function of the final output size. To reduce the footprint of each run, intermediate data are compressed. For example, for H-BNLJ, the size of intermediate data is 2. 6 times bigger than the size of output data. Overall, the algorithms with the lowest space requirements are RankReduce and PGBJ.
Impact of k Changing the value of k can have a significant impact on the performance of some of the kNN algorithms. We experimented on a dataset of 2 * 10 5 records (only 5 * 10 4 for H-BNLJ for performance reasons) with values for k varying from 2 to 512. Results are shown in Figure  7  using a logarithmic scale on the x-axis. First, we observe a global increase in computing time (Figure  7a ) which matches the complexity analysis performed earlier. As k increases, the performance of H-zkNNJ, compared to the other advanced algorithms, decreases. This is due to the necessary replication of the z-values of S throughout the partitions to find enough candidates: the core computation is thus much more complex. Second, the algorithms can also be distinguished considering their disk usage, visible on Figure  7b.
The global tendency is that the ratio of intermediate data size over the final data size decreases. This means that for each algorithm the final data size grows faster than the intermediate data size. As a consequence, there is no particular algorithm that suffers from such a bottleneck at this point. PGBJ is the most efficient from this aspect. Its replication of data occurs independently of the number of selected neighbors. Thus, increasing k has a small impact on this algorithm, both in computing time and space requirements. On this figure, an interesting observation can also be made for H-zkNNJ. For k = 2, it has by far the largest disk usage but becomes similar to the others for larger values. This is because H-zkNNJ creates a lot of intermediate data (copies of the initial dataset, vectors for the space filling curve, sampling.
) irrespective of the value of k. As k increases, so does the output size, mitigating the impact of these intermediate data. Surprisingly, changing k has a different impact on the recall of the approximate kNN methods, as can be seen on Figure  7c. For RankReduce, increasing k has a negative impact on the recall which sharply decreases when k ≥ 64. This is because the window parameter (W ) of LSH was set at the beginning of the experiments to achieve the best performance for this particular dataset. However, it was not modified for various of k. Thus it became less optimal as k increased. This shows there is a link between global parameters such as k and parameters of the LSH process. When using H-zkNNJ, increasing k improves the precision: the probability to have incorrect points is reduced as there are more candidates in a single partition.