Navigation
-
Lenses
This content is ...
Endorsed by
(What does "Endorsed by" mean?)Also in these lenses
-
Recently Viewed
- This feature requires Javascript to be enabled.
An Introduction to Parallel Programming with MapReduce
Summary: MapReduce is a framework based on the Map and Reduce functions used in functional languages like Lisp. QT Concurrent provides a MapReduce implementation for parallel programming on multi-core processors. This module provides an introduction to parallel programming with MapReduce, using QT Concurrent.
MapReduce Overview
MapReduce is a framework designed by Google [1]. It is loosely based on the Map and Reduce programming constructs of functional languages like Lisp [2]. Google’s MapReduce is used in a distributed computing, scalable platform. It was designed for data-intensive applications which need to process huge amounts of data.
Various open source implementations exist. The most popular one is Apache’s Hadoop [3]. Hadoop is a scalable distributed computing platform that includes a file system (HDFS) to store massive data, and a Java MapReduce implementation to process that data.
Other MapReduce implementations exist. QT Concurrent has a MapReduce implementation for multi-core processors.
This module provides an introduction to parallel programming with MapReduce, using QT Concurrent.
Programming with MapReduce
When we decide to solve a programming problem using MapReduce, we provide an implementation for a mapper and for a reducer. N mappers are executed simultaneously, executing the same task for different input data. One or more reducers receive the output generated by the mappers, and apply a reducing phase yielding a final result.
A simple program to count word occurrences in a text corpus is described in [4]:
map(String input_key, String input_value):
// input_key: document name
// input_value: document contents
for each word w in input_value:
EmitIntermediate(w, "1");
reduce(String output_key, Iterator intermediate_values):
// output_key: a word
// output_values: a list of counts
int result = 0;
for each v in intermediate_values:
result += ParseInt(v);
Emit(AsString(result));
MapReduce pseudocode that counts word occurrences in a text corpus.
As shown in the code above, a map function receives an input key and an input value (<key, value>), and generates one or more intermediate output <key,value(s)> pairs. A reduce function receives intermediate keys and values that were the ouput by the mappers, processes them in some way, and generates one or more final key-value(s) pairs (<key,value(s)).
You can find a Hadoop implementation of the word count problem at [5], and a QT concurrent implementation in the QT Concurrent package (path: qtconcurrent/examples/wordcount) which can be checked out with subversion.
MapReduce in QT Concurrent
QT Concurrent [6] is a C++ library for multi-threaded applications. Among other things, it provides a MapReduce implementation for multi-core computers. The map function is called in parallel by multiple threads. The number of threads used in a program depends on the number of processor cores available.
Google’s original MapReduce runs in computer clusters. The data it processes is stored in a distributed file system called GFS (Google File System). To minimize I/O bottlenecks, mappers are usually executed in the same node where the data resides. Parallelism comes from multiple computers executing mappers (or reducers) at the same time.
On the other hand, QT Concurrent’s MapReduce implementation works on shared-memory systems. Parallelism comes from multiple threads executing mappers at the same time, on multiple processor cores.
To work with QT Concurrent’s MapReduce, you write a map function and a reduce function. You must also indicate the list of values (e.g., names of files, words, numbers, etc.) that you want to feed your mappers. To do this, use the following API:
QFuture<T> QtConcurrent::mappedReduced (
const Sequence & sequence,
MapFunction mapFunction,
ReduceFunction reduceFunction,
QtConcurrent::ReduceOptions reduceOptions = UnorderedReduce | SequentialReduce )
Structure QTConcurrent::mappedReduced function
For example:
QFuture<T> mappedReduced(theList, mapFunction, reduceFunction);
QFuture::mappedReduced function
Example: Determining if a (Big) Integer is a Probable Prime
An algorithm for finding a probable prime can be defined using the Miller-Rabin primality test, as follows:
let n be a very big odd number
check n for divisibility for all primes < 2000
choose m positive integers less than n
for each of these bases apply the Miller-Rabin test
The algorithm described above can be easily parallelized using MapReduce as follows:
- Step 1. Randomly generate an odd (big) integer n greater than 2.
- Step 2. Generate (or read a list of) all primes less than 2000.
- Step 3. MapReduce-Part1:
- Map function: Each mapper receives a distinct prime number pi, and checks if n is divisible by this number. A mapper emits (outputs) a 1 if n is divisible by pi or 0 otherwise.
- Reduce function: Counts the number of divisors. If the number of divisors is greater than zero, it is not a prime number, so the test is terminated.
- Step 4. Generate 100 random positive integers less than n.
- Step 5. MapReduce-Part2:
- Map function: Each mapper receives a different random number ri and applies the Miller-Rabin primality test. A mapper emits (outputs) a 1 if n does not pass the test, or a 0 if it passes the test.
- Reduce function: Counts the number of 1s emitted by the mappers. If the number of divisors is zero, n is a probable prime.
The next section shows a MapReduce implementation of the algorithm described above, using QT Concurrent. You can find a Hadoop implementation here.
A MapReduce Implementation Using QT Concurrent
The main program needs to call the mappedReduced function twice (steps 3 and 5 in the algorithm described earlier):
qDebug() << "\nJob: Divisibility for primes less than 2000";
qDebug() << "Starting Job";
time.start();
Counting final = mappedReduced(primes,mapperLess2000,reducerLess2000);
mapReduce2000Time = time.elapsed();
qDebug() << "End Job";
qDebug() << "MapReduce elapsed time: " << (mapReduce2000Time) << "msegs\n";
//Counter of exacts divisors equals to zero
if(final[exactDiv] == 0) {
de = Decomposition<2048>(number - 1);
qDebug() << "Job: Divisibility for random numbers less than number to evaluate";
qDebug() << "Starting Job";
time.start();
Counting finalRand = mappedReduced(randomPrimes,mapperRandom,reducerRandom);
mapReduceRandomTime = time.elapsed();
qDebug() << "End Job";
qDebug() << "MapReduce elapsed time: " << (mapReduceRandomTime) << "msegs\n";
//Counter of random numbers
if(finalRand[falseMiller] > 0) {
qDebug() << "Result: Non-prime";
} else
qDebug() << "Result: Probably prime";
} else qDebug() << "Result: Non-prime";
Main program
The Decomposition function used in the main program because number decomposition is needed before the Miller-Rabin test is performed. This means that the number minus one is represented as product of 2k and m, where m is odd.
Two MapReduce processes are implemented. The first MapReduce process, evaluates divisibility for all primes less than 2000. Each map function receives a prime as argument and the reduce function counts the number of divisors, as shown below:
Counting mapperLess2000(const unsigned long &prime) {
Counting rMap;
BigInt2048 myPrime(prime);
if( number % myPrime == 0 )
divisors << prime;
rMap[( number % myPrime == 0 )?exactDiv:inexactDiv];
return rMap;
}
void reducerLess2000(Counting &result, const Counting &w) {
QMapIterator<QString, int> i(w);
while(i.hasNext()) {
i.next();
result[i.key()]++;
}
}
MapReduce code to test for primality
The second MapReduce process is executed only if the random number was not divisible by any of the prime numbers less than 2000:
Counting mapperRandom(const BigInt2048 &prime) {
Counting rMap;
BigInt2048 z( getZ(prime,de.getM()).bits );
// This is the easy case; the first term in the sequence
// is correct, so we pass the test.
if(z == 1) {
rMap[trueMiller];
divisors << z;
return rMap;
} else {
for (int j = 0; j < de.getK(); j++) {
BigInt2048 zSquared = BigInt2048(newZ(z).bits);
if (zSquared == 1 && z == (number - 1) ) {
// We've passed the hard version of the Rabin Miller test.
rMap[trueMiller];
divisors << z;
return rMap;
}
z = zSquared;
}
rMap[falseMiller];
return rMap;
}
}
void reducerRandom(Counting &result, const Counting &w) {
QMapIterator<QString, int> i(w);
while(i.hasNext()) {
i.next();
result[i.key()]++;
}
}
MapReduce code that applies the Miller-Rabin test
Figure 1 shows the results of one run of the program.
 |
You can download the complete source of this example here.
References
- [1] Jeffrey Dean, Sanjay Ghemawat. “MapReduce: Simplified Data Processing on Large Clusters.” In Communications of the ACM, vol. 51, no. 1 (2008). pp. 107-113.
- [2] McCarthy, J. "Recursive functions of symbolic expressions and their computation by machine, Part I". In Communications of the ACM. Vol 3 (4), pp. 184-195. Apr. 1960. http://www-formal.stanford.edu/jmc/recursive.pdf
- [3] Apache Hadoop, http://hadoop.apache.org/, 03/17/2009
- [4] Aaron Kimball, Sierra Michels-Slettvet, Christophe Bisciglia, et al. , “Lecture 2 - MapReduce: Theory and Implementation”. Introduction to Problem Solving on Large Scale Clusters. Spring 2007. Available for download at: http://code.google.com/edu/submissions/uwspr2007_clustercourse/listing.html
- [5] Hadoop Wiki WordCount, http://wiki.apache.org/hadoop/WordCount, 24/09/2008
- [6] Qt Concurrent, http://labs.trolltech.com/page/Projects/Threads/QtConcurrent , 12/02/2008
Content actions
Give Feedback:
Download:
Add module to:
Footer
More about this module: Metadata | Version History
This work is licensed by Cristina Abad and Allan Avendaño under a Creative Commons Attribution License (CC-BY 3.0), and is an Open Educational Resource.
Last edited by Allan Avendaño on Mar 25, 2009 11:51 am GMT-5.