We consider the problem of processing K-Nearest Neighbor (KNN) queries over large datasets where the index is jointly maintained by a set of machines in a computing cluster. The proposed RankReduce approach uses locality sensitive hashing (LSH) together with a MapReduce implementation, which by design is a perfect match as the hashing principle of LSH can be smoothly integrated in the mapping phase of MapReduce. The LSH algorithm assigns similar objects to the same fragments in the distributed file system which enables a effective selection of potential candidate neighbors which get then reduced to the set of K-Nearest Neighbors. We address problems arising due to the different characteristics of MapReduce and LSH to achieve an efficient search process on the one hand and high LSH accuracy on the other hand. We discuss several pitfalls and detailed descriptions on how to circumvent these. We evaluate RankReduce using both synthetic data and a dataset obtained from Flickr.com demonstrating the suitability of the approach.
With the success of the Web 2.0 and the wide spread usage
of cell phones and digital cameras, millions of pictures are
being taken and uploaded to portals like Facebook or Flickr
every day1, accumulating to billions of images2. Searching
in these huge amounts of images becomes a challenging task.
While there is an increasing trend to use social annotations,
so-called tags, for image retrieval, next to the traditional
image search `a la Google/Bing/Yahoo which inspects the text
around the web site holding the picture, there is a vital need
to process similarity queries, where for a given query
picture, the K most similar pictures are returned based on low
level features such as color, texture, and shape [
LSDS-IR Workshop, July 2010. Geneva, Switzerland.
available, whereas tags are usually extremely scarce and text
around images can often be misleading. Approaches like the
work by Taneva et al. [
The increasing volume of high dimensional data, however,
poses novel problems to traditional indexing mechanisms
which usually assume an in-memory index or optimize for
local disk access. As a promising approach to process huge
amounts of data on a multitude of machines in a cluster,
MapReduce [
In this work, we propose RankReduce, an approach to
implement locality sensitive hashing (LSH) [
The remainder of the paper is structured as follows. Section 2 gives an overview of related work, Section 3 presents our framework and gives a brief introduction to LSH and MapReduce, Section 4 describes the way queries are processed, Section 5 presents an experimental evaluation, and Section 6 concludes the paper and gives an outlook on ongoing work.
Processing K-Nearest Neighbor queries in high
dimensional data has received a lot of attention by researchers
in recent years. When the dimensionality increases the
distance between the closest and the farthest neighbor decreases
rapidly, for most of the datasets [
MapReduce is a framework for efficient and fault tolerant
workload distribution in large clusters [
We address the problem of processing K-Nearest Neighbor queries in large datasets by implementing a distributed LSH based index within the MapReduce Framework.
An LSH based index uses locality sensitive hash functions for indexing data. The salient property of these functions is that they map, with high probability, similar objects (represented in the d -dimensional vector space) to the same hash bucket, i.e., related objects are more probable to have the same hash value than distant ones. The actual indexing builds several hash tables with different LSH functions to increase the probability of collision for close points. At query time, the KNN search is performed by hashing the query point to one bucket per hash table and then to rank all discovered objects in any of these buckets by their distance to the query point. The closest K points are returned as the final result.
In this work, we consider the family of LSH functions
based on p-stable distributions [
W c (1) where a is a d -dimensional vector whose elements are chosen independently from a p-stable distribution, W ∈ IR, and B is chosen uniformly from [0, W ]. Each hash function maps a d-dimensional data point onto the set of integers. With k such hash functions, the final result is a vector of length k of the form g(v) = (ha1,B1 (v), ..., hak,Bk (v)).
In order to achieve high search accuracy, multiple hash
tables need to be constructed. The work by Lv et al. [
For maintaining the hash tables over a set of machines in a
cluster, we employ MapReduce [
In the Hadoop implementation, the input data is grouped in so-called input splits (which often correspond to blocks in the distributed file system), and a number of so-called mapper processes call the Map function for each key-value pair in such an input split. A number of mappers can run concurrently on each node in the cluster, and the mapper processes are in addition distributed over all nodes in the cluster. Ideally, a mapper is run on the same node where the input block resides, but this is not always possible due to workload imbalance. Similarly, after all mappers have finished, dedicated reducer processes are run on nodes int the cluster. Each reducer handles a fraction of the output key space, copies those key-value pairs from all mappers’ outputs (in the so-called shuffle phase), sorts them by key, and feeds the to the Reduce function. The output of the reducers is usually considered the final result but can also be used as input for following MapReduce jobs.
Figure 1 shows an illustration of our LSH integration with MapReduce. Each hash table in the LSH index is mapped to one folder in HDFS. For each bucket in such a hash table, a corresponding file is created in this folder, where the file name is created by concatenating hash values into a string, with ’ ’ as separator. This mapping of buckets to HDFS files enables fast lookup at query time and ensures that only data that is to be probed is read from the HDFS. Placing the bucket in one file also enables block based sequential access to all vectors in one bucket, which is very important as the MapReduce framework is optimized for such block based rather than random access processing. Each of the buckets stores the complete feature vectors of all objects mapped to this bucket in a binary encoding.
Indexing of new feature vectors to the LSH index in HDFS is easily done by appending them to the end of the appropriate bucket file. This can also be done in parallel with query processing as long as different buckets are affected; as HDFS does not include a transaction mechanism, appending entries to buckets that are being queried would be possible, but with unclear semantics for running queries. As HDFS scales well with increasing cluster size, the resulting growth of the LSH index can easily be supported by adding more machines to the cluster.
While an LSH index stored in-memory has no limitation on the number of buckets, too many files in HDFS can downgrade its performance, especially if these files are much smaller than the block size (which defaults to 64MB). The number of buckets, and therefore the number of files in HDFS for the LSH index, is highly dependent on the set up of LSH parameters as choosing a bad combination of parameters can result in a large number of small files.
Inspired by in-memory indexes which can have references from buckets to materialized feature vectors, we considered storing only feature vector ids in the buckets instead of the actual feature vectors, and retrieving the full vectors only on demand at query time. However, this approach would result in poor performance due to many random accesses to the HDFS when retrieving the full vectors, so we decided to store complete feature vectors. This fact also needs to be addressed when setting up LSH parameters, while too many LSH hash tables can dramatically increase index size, as each feature vector is materialized for each hash table. 4.
We implemented KNN query processing as a MapReduce job. Before starting this MapReduce job, the hash values for the query documents are calculated. These values are then used for selecting the buckets from the LSH index, which are to be probed. The selected buckets are provided as input to the query processing MapReduce job, generating multiple input splits. The generated input splits are read by a custom implementation of the InputFormat class, which reads feature vectors stored in a binary format and provides them as the key part of the Map function input. Queries are being distributed to mappers either by putting them in the Distributed Cache or by putting them in HDFS file with high number of replicas. They are read once by the InputFormat implementation and reused as value part of the Map function input between the function invocations.
The input to the Map function consists therefore of the feature vector to be probed as the key and the list of queries as the value. The Map function computes the similarity of the feature vector with all query vectors. While a standard MapReduce implementation would now emit a result pair for each combination of feature vector and query vector, we employ an optimization that delays emitting results until all feature vectors in the input split have been processed. We then eventually emit the final K-Nearest Neighbor for each query vector from this input split in the form of key-value pairs. Here, the query is the key and a nearest neighbor together with its distance to the query vector is the value. To implement this delayed emitting, we store the currently best K-Nearest Neighbor for each query in-memory, together with their distances from the query points. The results are emitted at the end of processing the input split in Hadoop’s cleanup method4. The Reduce method then reads, for each query, the K-Nearest Neighbor from each mapper, sorts them by increasing distance, and emits the best K of them as the final result for this query.
The final sort in the reducer can even be executed within Hadoop instead of inside the Reduce method, as a subtask of sorting keys in the reducer. It is possible to apply a so-called Secondary Sort that allows, in our application, to sort not just the keys, but also the values for the same key. Technically, this is implemented by replacing, for each (query, (neighbor, distance)) tuple that is emitted by a mapper, the key by a combined key consisting of the query and the distance. Keys are then sorted lexicographically first by query and then by distance. For assigning tuples to a Reduce method, however, only the query part of the key is taken into account. The reducer then only needs to read the first K values for each key, which then correspond to the K-Nearest Neighbor for that query.
It is worth mentioning that because one feature vector is placed in multiple hash tables, the same vector can be evaluated twice for the same query during processing. An 4This feature was introduced in the most recent version 0.20; before, it was only possible to emit directly from the Map function alternative approach would be to have two MapReduce jobs for query processing instead of one, which would eliminate this kind of redundancy. The first MapReduce job would create a union between buckets that need to be probed, and the second job would use the union as an input to similarity search. However, while this would possibly save redundant computations, it has the major drawback that the results from the first job need to be written to the HDFS before starting the second job. As the overhead from multiple evaluations of the same feature vector has not been too large in our experimental evaluation (see Figure 4), we decided that it is better to probe slightly more data rather than to pay the additional I/O cost incurred by using two Map Reduce jobs.
The approach can handle multiple queries at the same
time in one MapReduce job. But it is not suitable for the
cases when the number of queries becomes too large, as
problem of KNN queries processing becomes the problem of set
similarity joins [
For our experiments we have used Hadoop version 0.20.2 installed on three virtual machines with Debian GNU/Linux 5.0 (Kernel version: 2.6.30.10.1) as operating system. Each of the virtual machines has been configured to have 200GB hard drive, 5 GB main memory and two processors. VMware Server version 2.0.2 was used for virtualization of all machines. The virtual machines were run on a single machine with Intel Xeon CPU E5530 @2.4 GHz, 48 GB main memory, 4 TB of hard drive and Microsoft Windows Server 2008 R2 x64 as operating system. We used a single machine Hadoop installation on these virtual machines as described later on.
As the performance of the LSH based index is highly
dependent on the data characteristics [
Synthetic Dataset: For the synthetic dataset we used 32-dimensional randomly generated vectors. The synthetic dataset was built by first creating N independently generated vector instances drawn from the normal distribution N (0, 1) (independently for each dimension). Subsequently, we created m near duplicates for each of the N vectors, leading to an overall dataset size of m∗N vectors. The rational behind using the near duplicates is that we make sure that the KNN retrieval is meaninful at all. We set m to 10 in the experiments and adapt N to the desired dataset size depending on the experiment. We generated 50 queries by using the same procedure as the original vectors were generated.
Flickr Dataset:
We used the 64-dimensional color structure feature vectors
from crawled Flickr images provided by the CoPhIR data
collection [
As LSH is an approximate method, we measure the effectiveness of the nearest neighbor search by its precision, which is the relative overlap of the true K-Nearest Neighbor with the K-Nearest Neighbor computed by our method.
And for the proximity measure we used Euclidean distance. 5.1
Before starting the evaluation we needed to understand
how to set up LSH and what consequence it may have on
the index size and query performance. In our setup we
consider the number of hash tables and the bucket size as LSH
parameters to be tuned. The bucket size can be changed
either by changing the number of concatenated hash values or
by changing the parameter W in Formula 1. Because W is a
continuous variable and provides a subtle control over bucket
size, we first fix the number of concatenated hash values and
then vary parameter W [
Realizing that each new table creates another copy of data and we may have only limited storage available, we need to tradeoff storage cost vs. execution time. Additionally, when only a fixed subset of the data should be accessed, a larger number of hash tables results in a large number of small sized buckets, which is not a good scenario for HDFS (it puts additional pressure on Hadoop’s data node that manages all files). On one hand, we would like to increase the number of hash tables and to decrease the probed data subset. On the other hand, we would like to use less storage space and a smaller number of files for storage and probing. Figure 3 shows that the number of hash tables has smaller impact on precision in case of real image data. Thus, as a general rule we suggest a smaller number of hash tables with larger bucket sizes, still set to satisfy the precision threshold. Therefore we settle for a setup of four hash tables and a bucket size that allow us to get at least 70% precision. 5.2
We evaluate our approach and compare it to the linear
scan over all data, also implemented as a MapReduce job.
As we did not have a real compute cluster at hand for
running the experiments, we simulate the execution in a large
cluster by running the mappers and reducers sequentially
on our small machine. We measure run times of their
executions and the number of mappers started for each query
job. To avoid the possible bottleneck of a shared hard drive
between virtual machines [
Considering that the workload for the reducers is really small for both linear scan and LSH, we only evaluate map execution times and the number of mappers run per query job. We measured the map execution times for all jobs and found that they are approximately constant, with average value per mapper being 3.256 seconds and standard deviaLSH characteristics [Generated Data] LSH characteristics [Real Image Data] tion of 1.702 seconds. Taking into account that each mapper has an approximately same data input size, defined by the HDFS’ block size, approximately constant mapper execution time is well expected.
Because the execution time of the mappers is almost constant, the load of a query execution can be represented as number of mappers per query. We measured the number of mappers per query and precision for 50 KNN queries, with K=20, for both datasets, with 2GB, 4GB, and 8GB of indexed data (∼4000, ∼8000, and ∼16000 feature vectors for the real image dataset and ∼8000, ∼16000, and ∼32000 feature vectors for the synthetic dataset, respectively). The number of mappers per query for synthetic dataset is shown in Figure 5. And as we can see the number of mappers is about 3 times smaller for LSH than for linear scan. Also we can see in Figure 6, which shows the number of mappers per query for the Flickr dataset, that the difference in the number of mappers between LSH and linear scan is even bigger. The number of mappers per query is 4 to 5 times smaller for LSH than for linear scan in this case. The precision, shown in Figure 7, for generated data is over the threshold of 70% for 2GB and 4GB of indexed data, but drops down to 63.8% for 8GB. For real image data, the precision is almost constant, varying slightly around 86%.
In this work we described RankReduce, an approach for processing large amounts of data for K-Nearest Neighbor (KNN) queries. Instead of dealing with standard issues in distributed systems such as scalability and fault tolerance, we implement our solution with MapReduce, which provides these salient properties out of the box. The key idea of the presented approach is to use Locality Sensitive Hashing (LSH) in the Map phase of MapReduce to assign similar objects to the same files in the underlying distributed file system. While this seemed to be straight forward at first glance, there was a nontrivial conflict of opposing criteria and constraints caused by LSH and MapReduce which we had to solve to achieve accurate results with an acceptable query response time. We have demonstrated the suitability of our approach using both a synthetic and a real world dataset. 100 90 80
Our presented approach on large scale data processing is, however, not limited to KNN search over images, but can be extended to a variety of other interesting applications, such as near duplicate detection, document classification, or document clustering.
As a first step in our future work plan to evaluate our approach on a real compute cluster, which we are currently building up, with large scale data in the order of several TB. We furthermore plan to extend our approach to video and music retrieval. Map tasks per query [Generated Data] LSH precision [Generated and Real Image Data] 140 120
0 140 120 0
LSH Linear scan 2
4 Indexed data size GB
8 Map tasks per query [Real Image Data]
LSH
Linear scan Generated data
Picture data 4 5 6 7
Indexed data size GB Figure 7: LSH precision on generated and picture data.