Bespin is a library that contains reference implementations of "big data" algorithms in MapReduce and Spark.
Build the package:
$ mvn clean package
Grab the data:
$ mkdir data
$ curl http://lintool.github.io/bespin-data/Shakespeare.txt > data/Shakespeare.txt
$ curl http://lintool.github.io/bespin-data/p2p-Gnutella08-adj.txt > data/p2p-Gnutella08-adj.txt
The datasets are stored in the Bespin data repo.
- The file
Shakespeare.txtcontains the The Complete Works of William Shakespeare from Project Gutenberg. - The file
p2p-Gnutella08-adj.txtcontains a snapshot of the Gnutella peer-to-peer file sharing network from August 2002, where nodes represent hosts in the Gnutella network topology and edges represent connections between the Gnutella hosts. This dataset is available from the Stanford Network Analysis Project. - The tarball
taxi-data.tar.gzcontains a one-day slice NY taxi data, chopped into one file per minute. See analyses in Todd Schneider's blog post Analyzing 1.1 Billion NYC Taxi and Uber Trips, with a Vengeance.
Make sure you've downloaded the Shakespeare collection (see "Getting Started" above). Running word count in Java MapReduce:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.wordcount.WordCount \
-input data/Shakespeare.txt -output wc-jmr-combiner
To enable the "in-mapper combining" optimization, use the -imc option.
Running word count in Scala MapReduce:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.scala.mapreduce.wordcount.WordCount \
--input data/Shakespeare.txt --output wc-smr-combiner
To enable the "in-mapper combining" optimization, use the --imc option.
And finally, running word count in Spark:
$ spark-submit --class io.bespin.scala.spark.wordcount.WordCount target/bespin-1.0.5-SNAPSHOT-fatjar.jar \
--input data/Shakespeare.txt --output wc-spark-default
To enable the "in-mapper combining" optimization in Spark, use the --imc option (although this optimization doesn't do anything: exercise left to the reader as to why).
Compare results to make sure they are the same:
$ hadoop fs -cat wc-jmr-combiner/part-r-0000* | awk '{print $1,$2;}' | sort > counts.jmr.combiner.txt
$ hadoop fs -cat wc-smr-combiner/part-r-0000* | awk '{print $1,$2;}' | sort > counts.smr.combiner.txt
$ hadoop fs -cat wc-spark-default/part-0000* | sed -E 's/^\((.*),([0-9]+)\)$/\1 \2/' | sort > counts.spark.default.txt
$ diff counts.jmr.combiner.txt counts.smr.combiner.txt
$ diff counts.jmr.combiner.txt counts.spark.default.txt
Tip: We use awk in some cases and sed in others because sed does not accept control characters such as \t (but GNU sed does), so you have to insert a literal tab in the command line. This is awkward: on Mac OS X, you need to type ^V^I; so it's just easier with awk.
Make sure you've downloaded the Shakespeare collection (see "Getting Started" above). Running a simple bigram count:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bigram.BigramCount \
-input data/Shakespeare.txt -output bigram-count
Computing bigram relative frequencies using the "pairs" implementation:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bigram.ComputeBigramRelativeFrequencyPairs \
-input data/Shakespeare.txt -output bigram-freq-mr-pairs -textOutput
Computing bigram relative frequencies using the "stripes" implementation:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bigram.ComputeBigramRelativeFrequencyStripes \
-input data/Shakespeare.txt -output bigram-freq-mr-stripes -textOutput
To obtain human-readable output, make sure to use the -textOutput option; otherwise, the job defaults to SequenceFile output.
Let's spot-check the output to make sure the results are correct. For example, what are the bigrams that begin with "dream"?
$ hadoop fs -cat bigram-count/part* | grep '^dream '
dream again 2
dream and 2
dream are 1
dream as 1
dream away 2
...
What is the sum of all these counts?
$ hadoop fs -cat bigram-count/part* | grep '^dream ' | cut -f 2 | awk '{sum+=$1} END {print sum}'
79
Confirm that the numbers match the "pairs" implementation of the relative frequency computations:
$ hadoop fs -cat bigram-freq-mr-pairs/part* | grep '(dream, '
And the "stripes" implementation of the relative frequency computations:
$ hadoop fs -cat bigram-freq-mr-stripes/part* | awk '/^dream\t/'
Tip: Note that grep in Mac OS X accepts \t, but not on Linux; strictly speaking, grep uses regular expressions as defined by POSIX, and for whatever reasons POSIX does not define \t as tab. One workaround is to use -P, which specifies Perl regular expressions; however the -P option does not exist in Mac OS X.
Here's how you can verify that the pairs and stripes implementation give you the same results:
$ hadoop fs -cat bigram-freq-mr-pairs/part-r-0000* | awk '{print $1$2,$3;}' | grep -v ",\*)" | sort > freq.mr.pairs.txt
$ hadoop fs -cat bigram-freq-mr-stripes/part-r-0000* | perl -ne '%H=();m/([^\t]+)\t\{(.*)\}/; $k=$1; @k=split ", ",$2; foreach(@k){@p=split "=",$_;$H{$p[0]}=$p[1];}; foreach (sort keys %H) {print "($k,$_) $H{$_}\n";}' | sort > freq.mr.stripes.txt
$ diff freq.mr.stripes.txt freq.mr.pairs.txt
Make sure you've downloaded the Shakespeare collection (see "Getting Started" above). Running the "pairs" implementation:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.cooccur.ComputeCooccurrenceMatrixPairs \
-input data/Shakespeare.txt -output cooccur-pairs -window 2
Running the "stripes" implementation:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.cooccur.ComputeCooccurrenceMatrixStripes \
-input data/Shakespeare.txt -output cooccur-stripes -window 2
Let's spot check the results. For example, here are all the terms the co-occur with "dream" with the "pairs" implementation:
$ hadoop fs -cat cooccur-pairs/part* | grep '(dream, '
We can verify that the "stripes" implementation gives the same results.
$ hadoop fs -cat cooccur-stripes/part* | awk '/^dream\t/'
Tip: Note that grep in Mac OS X accepts \t, but not on Linux; strictly speaking, grep uses regular expressions as defined by POSIX, and for whatever reasons POSIX does not define \t as tab. One workaround is to use -P, which specifies Perl regular expressions; however the -P option does not exist in Mac OS X.
Make sure you've downloaded the Shakespeare collection (see "Getting Started" above). Building the inverted index:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.search.BuildInvertedIndex \
-input data/Shakespeare.txt -output index
Looking up an individual postings list:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.search.LookupPostings \
-index index -collection data/Shakespeare.txt -term "star-cross'd"
Running a boolean retrieval:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.search.BooleanRetrieval \
-index index -collection data/Shakespeare.txt -query "white red OR rose AND pluck AND"
Note that the query must be in Reverse Polish notation, so the above is equivalent to (white OR red) AND rose AND pluck in standard infix notation.
Make sure you've grabbed the sample graph data (see "Getting Started" above). First, convert the plain-text adjacency list representation into Hadoop Writable records:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bfs.EncodeBfsGraph \
-input data/p2p-Gnutella08-adj.txt -output graph-BFS/iter0000 -src 367
In the current implementation, you have to run a MapReduce job for every iteration, like this:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bfs.IterateBfs \
-input graph-BFS/iter0000 -output graph-BFS/iter0001 -partitions 5
Here's a bash script to run a bunch of iterations:
#!/bin/bash
for i in `seq 0 14`; do
cur=`echo $i | awk '{printf "%04d\n", $0;}'`
next=`echo $(($i+1)) | awk '{printf "%04d\n", $0;}'`
echo "Iteration $i: reading graph-BFS/iter$cur, writing: graph-BFS/iter$next"
hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bfs.IterateBfs -input "graph-BFS/iter$cur" -output "graph-BFS/iter$next" -partitions 5
done
The MapReduce job counters tell you how many nodes are reachable at each iteration:
| Iteration | Reachable | Distance |
|---|---|---|
| 0 | 1 | 1 |
| 1 | 9 | 8 |
| 2 | 65 | 56 |
| 3 | 257 | 192 |
| 4 | 808 | 551 |
| 5 | 1934 | 1126 |
| 6 | 3479 | 1545 |
| 7 | 4790 | 1311 |
| 8 | 5444 | 654 |
| 9 | 5797 | 353 |
| 10 | 5920 | 123 |
| 11 | 5990 | 70 |
| 12 | 6018 | 28 |
| 13 | 6026 | 8 |
| 14 | 6028 | 2 |
| 15 | 6028 | 0 |
To find all the nodes that are reachable at a particular iteration, run the following job:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bfs.FindReachableNodes \
-input graph-BFS/iter0005 -output graph-BFS/reachable-iter0005
$ hadoop fs -cat 'graph-BFS/reachable-iter0005/part*' | wc
These values should be the same as those in the second column of the table above.
To find all the nodes that are at a particular distance (e.g., the search frontier), run the following job:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bfs.FindNodeAtDistance \
-input graph-BFS/iter0005 -output graph-BFS/d0005 -distance 5
$ hadoop fs -cat 'graph-BFS/d0005/part*' | wc
The results are shown in the third column of the table above.
Here's a simple bash script for iterating through the reachability jobs:
#!/bin/bash
for i in `seq 0 15`; do
cur=`echo $i | awk '{printf "%04d\n", $0;}'`
echo "Iteration $i: reading graph-BFS/iter$cur"
hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bfs.FindReachableNodes -input graph-BFS/iter$cur -output graph-BFS/reachable-iter$cur
done
Here's a simple bash script for extracting nodes at each distance:
#!/bin/bash
for i in `seq 0 15`; do
cur=`echo $i | awk '{printf "%04d\n", $0;}'`
echo "Iteration $i: reading graph-BFS/iter$cur"
hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.bfs.FindNodeAtDistance -input graph-BFS/iter$cur -output graph-BFS/d$cur -distance $i
done
Make sure you've grabbed the sample graph data (see "Getting Started" above). First, convert the plain-text adjacency list representation into Hadoop Writable records:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.pagerank.BuildPageRankRecords \
-input data/p2p-Gnutella08-adj.txt -output graph-PageRankRecords -numNodes 6301
Create the directory where the graph is going to go:
$ hadoop fs -mkdir graph-PageRank
Partition the graph:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.pagerank.PartitionGraph \
-input graph-PageRankRecords -output graph-PageRank/iter0000 -numPartitions 5 -numNodes 6301
Run 15 iterations:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.pagerank.RunPageRankBasic \
-base graph-PageRank -numNodes 6301 -start 0 -end 15 -useCombiner
Extract the top 20 nodes by PageRank value and examine the results:
$ hadoop jar target/bespin-1.0.5-SNAPSHOT-fatjar.jar io.bespin.java.mapreduce.pagerank.FindMaxPageRankNodes \
-input graph-PageRank/iter0015 -output graph-PageRank-top20 -top 20
$ hadoop fs -cat graph-PageRank-top20/part-r-00000
367 -6.03734
249 -6.12637
145 -6.18742
264 -6.21511
266 -6.23297
123 -6.28525
127 -6.28685
122 -6.29073
1317 -6.29597
5 -6.30274
251 -6.32983
427 -6.33821
149 -6.40216
176 -6.42350
353 -6.43988
390 -6.44404
559 -6.45491
124 -6.45705
4 -6.47055
7 -6.50145
Compare the results with a sequential PageRank implementation:
$ mvn exec:java -Dexec.mainClass=io.bespin.java.mapreduce.pagerank.SequentialPageRank \
-Dexec.args="-input data/p2p-Gnutella08-adj.txt -jump 0.15"
The results should be the same.