In chapter 5 of the book "Data-Intensive Text Processing with MapReduce", it introduced how to parallel breadth-first graph search with MapReduce. This parallel algorithm is a variant of Dijkstra's algorithm. I'm not going to talk about the sequential version of Dijkstra's algorithm, for detailed explaination, refer to wikipedia. Also I'm not going to talk about the Graph data structure, please refer to wikipedia too.
If you know Dijkstra's algorithn well, you will know that the key to to Dijkstra's algorithm is the priority queue that maintains a globally sorted list of nodes by current distance. This is not possible in MapReduce, as the programming model does not provide a mechanism for exchanging global data. Instead, we adopt a brute force approach known as parallel breadth-first search.
The algorithm works by mapping over all nodes and emitting a key-value pair for each neighbor on the node's adjacency list. The key contains the node id of the neighbor, and the value is the current distance to the node plus the distance between current node and the next node. If we can reach node n with a distance d, then we must be able to reach all the nodes that we connected to n with distance d + dn. After shuffle and sort, reducers will receive keys corresponding to the destination node ids and distances corresponding to all paths leading to that node. The reducer will select the shortest of these distances and then update the distance in the node data structure.
It is apprant that parallel breadth-first search is an interative algorithm, where each iteration corresponds to a MapReduce Job. With each iteration, we will discover all nodes that are connected to current node. Also we need to pass along the graph structure from one iteration to the next. This is accomplished by emitting the node data structure itself in mapper. In the reducer, we must distinguish the node data structure from distance values and update the min distance in the node data structure before emitting it as the final value. Now it's ready to serve as input to the next iteration.
But the problem is, how may iterations are necessary to compute the shortest distance to all nodes? The answer is the dismeter of the graph, or the greatest distance between any pair of nodes. The algorithm can terminate when shortest distances at every node no longer change. We can use counters to keep track of such events. At the end of each MapReduce iteration, the driver program reads the counter value and determines if another is necessary.
package graph;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;
public class ParallelDijkstra extends Configured implements Tool {
public static String OUT = "output";
public static String IN = "inputlarger";
public static class DijkstraMapper extends Mapper<LongWritable, Text, LongWritable, Text> {
public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//From slide 20 of Graph Algorithms with MapReduce (by Jimmy Lin, Univ @ Maryland)
//Key is node n
//Value is D, Points-To
//For every point (or key), look at everything it points to.
//Emit or write to the points to variable with the current distance + 1
Text word = new Text();
String line = value.toString();//looks like 1 0 2:3:
String[] sp = line.split(" ");//splits on space
int distanceAdded = Integer.parseInt(sp[1]) + 1;
String[] pointsTo = sp[2].split(":");
for (String distance : pointsTo) {
word.set("VALUE " + distanceAdded);//tells me to look at distance value
context.write(new LongWritable(Integer.parseInt(distance)), word);
word.clear();
}
//pass in current node's distance (if it is the lowest distance)
word.set("VALUE " + sp[1]);
context.write(new LongWritable(Integer.parseInt(sp[0])), word);
word.clear();
word.set("NODES " + sp[2]);//tells me to append on the final tally
context.write(new LongWritable(Integer.parseInt(sp[0])), word);
word.clear();
}
}
public static class DijkstraReducer extends Reducer<LongWritable, Text, LongWritable, Text> {
public void reduce(LongWritable key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
//From slide 20 of Graph Algorithms with MapReduce (by Jimmy Lin, Univ @ Maryland)
//The key is the current point
//The values are all the possible distances to this point
//we simply emit the point and the minimum distance value
String nodes = "UNMODED";
Text word = new Text();
int lowest = 10009;//start at infinity
for (Text val : values) {//looks like NODES/VALUES 1 0 2:3:, we need to use the first as a key
String[] sp = val.toString().split(" ");//splits on space
//look at first value
if (sp[0].equalsIgnoreCase("NODES")) {
nodes = null;
nodes = sp[1];
} else if (sp[0].equalsIgnoreCase("VALUE")) {
int distance = Integer.parseInt(sp[1]);
lowest = Math.min(distance, lowest);
}
}
word.set(lowest + " " + nodes);
context.write(key, word);
word.clear();
}
}
//Almost exactly from http://hadoop.apache.org/mapreduce/docs/current/mapred_tutorial.html
public int run(String[] args) throws Exception {
//http://code.google.com/p/joycrawler/source/browse/NetflixChallenge/src/org/niubility/learning/knn/KNNDriver.java?r=242
//make the key -> value space separated (for iterations)
getConf().set("mapred.textoutputformat.separator", " ");
//set in and out to args.
IN = args[0];
OUT = args[1];
String infile = IN;
String outputfile = OUT + System.nanoTime();
boolean isdone = false;
boolean success = false;
HashMap<Integer, Integer> _map = new HashMap<Integer, Integer>();
while (!isdone) {
Job job = new Job(getConf(), "Dijkstra");
job.setJarByClass(ParallelDijkstra.class);
job.setOutputKeyClass(LongWritable.class);
job.setOutputValueClass(Text.class);
job.setMapperClass(DijkstraMapper.class);
job.setReducerClass(DijkstraReducer.class);
job.setInputFormatClass(TextInputFormat.class);
job.setOutputFormatClass(TextOutputFormat.class);
FileInputFormat.addInputPath(job, new Path(infile));
FileOutputFormat.setOutputPath(job, new Path(outputfile));
success = job.waitForCompletion(true);
//remove the input file
//http://eclipse.sys-con.com/node/1287801/mobile
if (!infile.equals(IN)) {
String indir = infile.replace("part-r-00000", "");
Path ddir = new Path(indir);
FileSystem dfs = FileSystem.get(getConf());
dfs.delete(ddir, true);
}
infile = outputfile + "/part-r-00000";
outputfile = OUT + System.nanoTime();
//do we need to re-run the job with the new input file??
//http://www.hadoop-blog.com/2010/11/how-to-read-file-from-hdfs-in-hadoop.html
isdone = true;//set the job to NOT run again!
Path ofile = new Path(infile);
FileSystem fs = FileSystem.get(new Configuration());
BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(ofile)));
HashMap<Integer, Integer> imap = new HashMap<Integer, Integer>();
String line = br.readLine();
while (line != null) {
//each line looks like 0 1 2:3:
//we need to verify node -> distance doesn't change
String[] sp = line.split(" ");
int node = Integer.parseInt(sp[0]);
int distance = Integer.parseInt(sp[1]);
imap.put(node, distance);
line = br.readLine();
}
if (_map.isEmpty()) {
//first iteration... must do a second iteration regardless!
isdone = false;
} else {
//http://www.java-examples.com/iterate-through-values-java-hashmap-example
//http://www.javabeat.net/articles/33-generics-in-java-50-1.html
for (Integer key : imap.keySet()) {
int val = imap.get(key);
if (_map.get(key) != val) {
//values aren't the same... we aren't at convergence yet
isdone = false;
}
}
}
if (!isdone) {
_map.putAll(imap);//copy imap to _map for the next iteration (if required)
}
}
return success ? 0 : 1;
}
public static void main(String[] args) throws Exception {
System.exit(ToolRunner.run(new ParallelDijkstra(), args));
}
}
The code is quoted from this blog post, also you can read the blog post for the instruction of how to run the code in Eclipse IDE with Hadoop plugin. In my case, I will run it in a tiny Hadoop cluster.
The input data:
1 0 2:3: 2 10000 1:4:5: 3 10000 1: 4 10000 2:5: 5 10000 2:4:
Output:
[root@n1 hadoop-examples]# hadoop jar hadoop-algorithms.jar graph.ParallelDijkstra inputgraph output 13/07/13 20:07:51 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. 13/07/13 20:08:03 INFO input.FileInputFormat: Total input paths to process : 1 13/07/13 20:09:04 INFO mapred.JobClient: Running job: job_201307131656_0001 13/07/13 20:09:05 INFO mapred.JobClient: map 0% reduce 0% 13/07/13 20:10:05 INFO mapred.JobClient: map 100% reduce 0% 13/07/13 20:15:13 INFO mapred.JobClient: map 100% reduce 100% 13/07/13 20:15:16 INFO mapred.JobClient: Job complete: job_201307131656_0001 13/07/13 20:15:16 INFO mapred.JobClient: Counters: 32 13/07/13 20:15:16 INFO mapred.JobClient: File System Counters 13/07/13 20:15:16 INFO mapred.JobClient: FILE: Number of bytes read=183 13/07/13 20:15:16 INFO mapred.JobClient: FILE: Number of bytes written=313855 13/07/13 20:15:16 INFO mapred.JobClient: FILE: Number of read operations=0 13/07/13 20:15:16 INFO mapred.JobClient: FILE: Number of large read operations=0 13/07/13 20:15:16 INFO mapred.JobClient: FILE: Number of write operations=0 13/07/13 20:15:16 INFO mapred.JobClient: HDFS: Number of bytes read=173 13/07/13 20:15:16 INFO mapred.JobClient: HDFS: Number of bytes written=53 13/07/13 20:15:16 INFO mapred.JobClient: HDFS: Number of read operations=2 13/07/13 20:15:16 INFO mapred.JobClient: HDFS: Number of large read operations=0 13/07/13 20:15:16 INFO mapred.JobClient: HDFS: Number of write operations=1 13/07/13 20:15:16 INFO mapred.JobClient: Job Counters 13/07/13 20:15:16 INFO mapred.JobClient: Launched map tasks=1 13/07/13 20:15:16 INFO mapred.JobClient: Launched reduce tasks=1 13/07/13 20:15:16 INFO mapred.JobClient: Data-local map tasks=1 13/07/13 20:15:16 INFO mapred.JobClient: Total time spent by all maps in occupied slots (ms)=38337 13/07/13 20:15:16 INFO mapred.JobClient: Total time spent by all reduces in occupied slots (ms)=159310 13/07/13 20:15:16 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0 13/07/13 20:15:16 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0 13/07/13 20:15:16 INFO mapred.JobClient: Map-Reduce Framework 13/07/13 20:15:16 INFO mapred.JobClient: Map input records=5 13/07/13 20:15:16 INFO mapred.JobClient: Map output records=20 13/07/13 20:15:16 INFO mapred.JobClient: Map output bytes=383 13/07/13 20:15:16 INFO mapred.JobClient: Input split bytes=112 13/07/13 20:15:16 INFO mapred.JobClient: Combine input records=0 13/07/13 20:15:16 INFO mapred.JobClient: Combine output records=0 13/07/13 20:15:16 INFO mapred.JobClient: Reduce input groups=5 13/07/13 20:15:16 INFO mapred.JobClient: Reduce shuffle bytes=179 13/07/13 20:15:16 INFO mapred.JobClient: Reduce input records=20 13/07/13 20:15:16 INFO mapred.JobClient: Reduce output records=5 13/07/13 20:15:16 INFO mapred.JobClient: Spilled Records=40 13/07/13 20:15:16 INFO mapred.JobClient: CPU time spent (ms)=3970 13/07/13 20:15:16 INFO mapred.JobClient: Physical memory (bytes) snapshot=240209920 13/07/13 20:15:16 INFO mapred.JobClient: Virtual memory (bytes) snapshot=2276737024 13/07/13 20:15:16 INFO mapred.JobClient: Total committed heap usage (bytes)=101519360 13/07/13 20:15:17 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. 13/07/13 20:15:18 INFO input.FileInputFormat: Total input paths to process : 1 13/07/13 20:15:18 INFO mapred.JobClient: Running job: job_201307131656_0002 13/07/13 20:15:19 INFO mapred.JobClient: map 0% reduce 0% 13/07/13 20:15:34 INFO mapred.JobClient: map 100% reduce 0% 13/07/13 20:15:41 INFO mapred.JobClient: map 100% reduce 100% 13/07/13 20:15:44 INFO mapred.JobClient: Job complete: job_201307131656_0002 13/07/13 20:15:44 INFO mapred.JobClient: Counters: 32 13/07/13 20:15:44 INFO mapred.JobClient: File System Counters 13/07/13 20:15:44 INFO mapred.JobClient: FILE: Number of bytes read=190 13/07/13 20:15:44 INFO mapred.JobClient: FILE: Number of bytes written=312999 13/07/13 20:15:44 INFO mapred.JobClient: FILE: Number of read operations=0 13/07/13 20:15:44 INFO mapred.JobClient: FILE: Number of large read operations=0 13/07/13 20:15:44 INFO mapred.JobClient: FILE: Number of write operations=0 13/07/13 20:15:44 INFO mapred.JobClient: HDFS: Number of bytes read=188 13/07/13 20:15:44 INFO mapred.JobClient: HDFS: Number of bytes written=45 13/07/13 20:15:44 INFO mapred.JobClient: HDFS: Number of read operations=2 13/07/13 20:15:44 INFO mapred.JobClient: HDFS: Number of large read operations=0 13/07/13 20:15:44 INFO mapred.JobClient: HDFS: Number of write operations=1 13/07/13 20:15:44 INFO mapred.JobClient: Job Counters 13/07/13 20:15:44 INFO mapred.JobClient: Launched map tasks=1 13/07/13 20:15:44 INFO mapred.JobClient: Launched reduce tasks=1 13/07/13 20:15:44 INFO mapred.JobClient: Data-local map tasks=1 13/07/13 20:15:44 INFO mapred.JobClient: Total time spent by all maps in occupied slots (ms)=16471 13/07/13 20:15:44 INFO mapred.JobClient: Total time spent by all reduces in occupied slots (ms)=6476 13/07/13 20:15:44 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0 13/07/13 20:15:44 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0 13/07/13 20:15:44 INFO mapred.JobClient: Map-Reduce Framework 13/07/13 20:15:44 INFO mapred.JobClient: Map input records=5 13/07/13 20:15:44 INFO mapred.JobClient: Map output records=20 13/07/13 20:15:44 INFO mapred.JobClient: Map output bytes=359 13/07/13 20:15:44 INFO mapred.JobClient: Input split bytes=135 13/07/13 20:15:44 INFO mapred.JobClient: Combine input records=0 13/07/13 20:15:44 INFO mapred.JobClient: Combine output records=0 13/07/13 20:15:44 INFO mapred.JobClient: Reduce input groups=5 13/07/13 20:15:44 INFO mapred.JobClient: Reduce shuffle bytes=186 13/07/13 20:15:44 INFO mapred.JobClient: Reduce input records=20 13/07/13 20:15:44 INFO mapred.JobClient: Reduce output records=5 13/07/13 20:15:44 INFO mapred.JobClient: Spilled Records=40 13/07/13 20:15:44 INFO mapred.JobClient: CPU time spent (ms)=2290 13/07/13 20:15:44 INFO mapred.JobClient: Physical memory (bytes) snapshot=250818560 13/07/13 20:15:44 INFO mapred.JobClient: Virtual memory (bytes) snapshot=1309839360 13/07/13 20:15:44 INFO mapred.JobClient: Total committed heap usage (bytes)=81399808 13/07/13 20:15:45 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same. 13/07/13 20:15:45 INFO input.FileInputFormat: Total input paths to process : 1 13/07/13 20:15:46 INFO mapred.JobClient: Running job: job_201307131656_0003 13/07/13 20:15:47 INFO mapred.JobClient: map 0% reduce 0% 13/07/13 20:15:56 INFO mapred.JobClient: map 100% reduce 0% 13/07/13 20:16:02 INFO mapred.JobClient: map 100% reduce 100% 13/07/13 20:16:04 INFO mapred.JobClient: Job complete: job_201307131656_0003 13/07/13 20:16:04 INFO mapred.JobClient: Counters: 32 13/07/13 20:16:04 INFO mapred.JobClient: File System Counters 13/07/13 20:16:04 INFO mapred.JobClient: FILE: Number of bytes read=172 13/07/13 20:16:04 INFO mapred.JobClient: FILE: Number of bytes written=312907 13/07/13 20:16:04 INFO mapred.JobClient: FILE: Number of read operations=0 13/07/13 20:16:04 INFO mapred.JobClient: FILE: Number of large read operations=0 13/07/13 20:16:04 INFO mapred.JobClient: FILE: Number of write operations=0 13/07/13 20:16:04 INFO mapred.JobClient: HDFS: Number of bytes read=180 13/07/13 20:16:04 INFO mapred.JobClient: HDFS: Number of bytes written=45 13/07/13 20:16:04 INFO mapred.JobClient: HDFS: Number of read operations=2 13/07/13 20:16:04 INFO mapred.JobClient: HDFS: Number of large read operations=0 13/07/13 20:16:04 INFO mapred.JobClient: HDFS: Number of write operations=1 13/07/13 20:16:04 INFO mapred.JobClient: Job Counters 13/07/13 20:16:04 INFO mapred.JobClient: Launched map tasks=1 13/07/13 20:16:04 INFO mapred.JobClient: Launched reduce tasks=1 13/07/13 20:16:04 INFO mapred.JobClient: Data-local map tasks=1 13/07/13 20:16:04 INFO mapred.JobClient: Total time spent by all maps in occupied slots (ms)=8584 13/07/13 20:16:04 INFO mapred.JobClient: Total time spent by all reduces in occupied slots (ms)=4759 13/07/13 20:16:04 INFO mapred.JobClient: Total time spent by all maps waiting after reserving slots (ms)=0 13/07/13 20:16:04 INFO mapred.JobClient: Total time spent by all reduces waiting after reserving slots (ms)=0 13/07/13 20:16:04 INFO mapred.JobClient: Map-Reduce Framework 13/07/13 20:16:04 INFO mapred.JobClient: Map input records=5 13/07/13 20:16:04 INFO mapred.JobClient: Map output records=20 13/07/13 20:16:04 INFO mapred.JobClient: Map output bytes=335 13/07/13 20:16:04 INFO mapred.JobClient: Input split bytes=135 13/07/13 20:16:04 INFO mapred.JobClient: Combine input records=0 13/07/13 20:16:04 INFO mapred.JobClient: Combine output records=0 13/07/13 20:16:04 INFO mapred.JobClient: Reduce input groups=5 13/07/13 20:16:04 INFO mapred.JobClient: Reduce shuffle bytes=168 13/07/13 20:16:04 INFO mapred.JobClient: Reduce input records=20 13/07/13 20:16:04 INFO mapred.JobClient: Reduce output records=5 13/07/13 20:16:04 INFO mapred.JobClient: Spilled Records=40 13/07/13 20:16:04 INFO mapred.JobClient: CPU time spent (ms)=1880 13/07/13 20:16:04 INFO mapred.JobClient: Physical memory (bytes) snapshot=286953472 13/07/13 20:16:04 INFO mapred.JobClient: Virtual memory (bytes) snapshot=3241136128 13/07/13 20:16:04 INFO mapred.JobClient: Total committed heap usage (bytes)=139132928
From the output we can see that the job iterates for three times with the given input graph data.
相关推荐
前端大厂最新面试题-breadth-first-search.docx
FastBFS: Fast Breadth-First Graph Search on a Single Server
广度优先搜索，BFS，breadth-first-search queue 队列，new 算法导论 ，Introduction to Algorithms是对算法导论上的广度优先搜索的实现，关键步骤都打印出了提示信息和当时的运行结果，VSCode有时候调试看不到变量...
matlab开发-BreadthFirstSearch。BFS使用源向量和目标向量对图进行广度优先搜索。
Improving vertex-frontier based GPU breadth-first search.pdf
其中，广度优先搜索（Breadth-First Search，简称BFS）和深度优先搜索（Depth-First Search，简称DFS）是两种最基本且广泛使用的图遍历算法。 广度优先搜索（BFS）是一种按层次遍历图的算法。它从图的某个顶点开始...
Breadth-First Search (BFS) is a key graph algorithm with many important applications. In this work, we focus on a special class of graph traversal algorithm - concurrent BFS - where multiple breadth-...
此类用于使用与“labeled_edges.tsv”相同的文件格式生成 SimpleGraph 对象 漫威主程序 这个类实际上从“labeled_edges.tsv”创建了大的 Marvel Graph。 它还多次调用 BFS 以向您展示它在所需的数据集上工作。 MP5....
city block breadth first search
广度优先搜索 在图中找到最短路径介绍代码给出了从源到目标的最短路径，输入必须以邻接表的形式给出，输出是点与点之间距离最短的路径。 0 1 4 1 0 4 2 3 2 1 3 3 1 4 2 4 3 0 1
5.2 Parallel Breadth-First Search 5.3 PageRank 5.4 Issues with Graph Processing 5.5 Summary and Additional Readings 6 EM Algorithms for Text Processing 6.1 Expectation Maximization 6.1.1 Maximum ...
广度优先搜索和迭代加深深度优先搜索 问题：使用不知情的搜索算法（广度优先搜索和迭代加深深度优先）解决15难题 初始状态：（简单的情况）1 2 4 5 7 3 8 9 6 11 12 13 10 14 15目标状态：1 2 3 4 5 6 7 8 9 10 11 ...
5.2 Parallel Breadth-First Search . . . . . . . . . . . . . . . . . . . 90 5.3 PageRank . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . 97 5.4 Issues with Graph Processing . . . . . . . . ...
Breadth First Search Java samples
广度优先搜索 解决问题的方法 经典的广度优先搜索算法的实现。 时间复杂度-O（V + E）（V-顶点数，E-边） Java版本：1.7
JS-递归广度优先搜索 具有递归广度优先搜索的二叉搜索树。 这是在 JavaScript 中创建具有递归广度优先搜索的二叉搜索树的尝试。 我将尝试使用两部分递归来实现这一点。 二叉搜索树(BST) 上的广度优先搜索(BFS) 在...
graphcut Matlab代码graph_TV_recond 此代码实现以下内容： 通过基于活动集的组合优化来优化图的总变化Ye，Z.，Möllenhoff，T.，Wu，T.，...breadth-first search}, author={Goldberg, Andrew V and Hed, Sagi a
通过实施BFS和React制作的社交网络推荐系统和联系查找器。
迷宫求解器深度和广度优先搜索示例 这是另一个学生和我为 Auburn 的 Java 课程创建的迷宫求解器程序。
8-1 terminology 8-2 graph storage structures 8-2-1 Adjacency Matrix ...8-4-2 Breadth-First Traversal 8-5 graph algorithms 8-6 networks 8-6-1 minimum spanning tree最小生成树 8-6-2 shortest path algorithm