示例：
给出各个节点到相邻节点的距离，要求算出初始节点到各个节点的最短路径。
数据：
A (B,10) (D,5) B (C,1) (D,2) C (E,4) D (B,3) (C,9) (E,2) E (A,7) (C,6)
A节点为初始节点，A到B的距离为10，A到D的距离为5。
B节点到C的距离为1，B到D的距离为2，其他类推。
MapReduce计算最短路径
Map阶段
如：
A (B,10) (D,5)
=>
A 0 (B,10) (D,5) #A到A的最短距离为0
B 10 #存在A到B的距离为10
D 5 #存在A到D的距离为5
从初始节点开始，将节点到其他相连节点的距离列举出来，然后传递给reduce，找到距离最短的。
记住从初始节点开始，从A开始，找到B和D，然后再找B和D的相邻节点，依次类推，这个就是广度优先搜索。
从A节点出发，A节点没有到达的节点默认的距离为inf表示无穷大。
Reduce阶段
找到所有存在的距离中最短的，并更新记录中的最短距离。
如：针对key值为B的
B inf (C,1) (D,2) #inf为最远距离，
B 10
=>
B 10 (C,1) (D,2) #A到B的最短距离为10
MapReduce过程中数据的变化：
原始数据：
A (B,10) (D,5) B (C,1) (D,2) C (E,4) D (B,3) (C,9) (E,2) E (A,7) (C,6) 第一次mr结果： A 0 (B,10) (D,5) #从初始节点A出发，找到A到B节点和D节点的距离 B 10 (C,1) (D,2) #找到B节点，且更新值，A到B节点目前的最短距离 C inf (E,4) D 5 (B,3) (C,9) (E,2) #找到D节点，且更新值，A到D节点目前的最短距离 E inf (A,7) (C,6) 第二次mr结果 A 0 (B,10) (D,5) B 8 (C,1) (D,2) C 11 (E,4) D 5 (B,3) (C,9) (E,2) E 7 (A,7) (C,6) 第三次mr结果 A 0 (B,10) (D,5) B 8 (C,1) (D,2) C 9 (E,4) D 5 (B,3) (C,9) (E,2) E 7 (A,7) (C,6) 第四次mr结果 A 0 (B,10) (D,5) B 8 (C,1) (D,2) C 9 (E,4) D 5 (B,3) (C,9) (E,2) E 7 (A,7) (C,6)
接下来还要考虑，什么时候所有节点的最短距离都计算完成？
我的计算方式，假设如果所有节点没有距离进行更新，说明所有节点的最短距离都已经计算完成，则完成计算。
源代码：
RunJob.java
1 import java.io.IOException; 2 3 import org.apache.hadoop.conf.Configuration; 4 import org.apache.hadoop.fs.FileSystem; 5 import org.apache.hadoop.fs.Path; 6 import org.apache.hadoop.io.Text; 7 import org.apache.hadoop.mapreduce.Job; 8 import org.apache.hadoop.mapreduce.Mapper; 9 import org.apache.hadoop.mapreduce.Reducer; 10 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 11 import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat; 12 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 13 import org.apache.hadoop.util.StringUtils; 14 15 16 /** 17 * Created by Edward on 2016/7/15. 18 */ 19 public class RunJob { 20 21 static enum eInf { 22 COUNTER 23 } 24 25 public static void main(String[] args) { 26 Configuration conf = new Configuration(); 27 28 conf.set("fs.defaultFS", "hdfs://node1:8020"); 29 30 try { 31 FileSystem fs = FileSystem.get(conf); 32 int i = 0; 33 long num = 1; 34 long tmp = 0; 35 while (num > 0) { 36 i++; 37 conf.setInt("run.counter", i); 38 Job job = Job.getInstance(conf); 39 job.setJarByClass(RunJob.class); 40 job.setMapperClass(ShortestPathMapper.class); 41 job.setReducerClass(ShortestPathReducer.class); 42 job.setMapOutputKeyClass(Text.class); 43 job.setMapOutputValueClass(Text.class); 44 45 //key value 的格式 第一个item为key，后面的item为value 46 job.setInputFormatClass(KeyValueTextInputFormat.class); 47 48 if (i == 1) 49 FileInputFormat.addInputPath(job, new Path("/test/shortestpath/input/")); 50 else 51 FileInputFormat.addInputPath(job, new Path("/test/shortestpath/output/sp" + (i - 1))); 52 53 Path outPath = new Path("/test/shortestpath/output/sp" + i); 54 if (fs.exists(outPath)) { 55 fs.delete(outPath, true); 56 } 57 FileOutputFormat.setOutputPath(job, outPath); 58 59 boolean b = job.waitForCompletion(true); 60 61 if (b) { 62 num = job.getCounters().findCounter(eInf.COUNTER).getValue(); 63 if (num == 0) { 64 System.out.println("执行了" + i + "次，完成最短路径的计算"); 65 } 66 } 67 } 68 69 } catch (Exception e) { 70 71 e.printStackTrace(); 72 } 73 74 } 75 76 /** 77 * @author Edward 78 * 79 * @1 A (B,10) (D,5) => 80 * A 0 (B,10) (D,5) 81 * B 10 82 * D 5 83 * @2 B 10 (C,1) (D,2) => 84 * B 10 (C,1) (D,2) 85 * C 11 86 * D 13 87 */ 88 public static class ShortestPathMapper extends Mapper<Text, Text, Text, Text> { 89 90 protected void map(Text key, Text value, Context context) throws IOException, InterruptedException { 91 int conuter = context.getConfiguration().getInt("run.counter", 1); 92 93 Node node = new Node(); 94 String distance = null; 95 String str = null; 96 97 // 第一次计算，填写默认距离 A:0 其他:inf 98 if (conuter == 1) { 99 if (key.toString().equals("A") || key.toString().equals("1")) { 100 distance = "0"; 101 } else { 102 distance = "inf"; 103 } 104 str = distance + "\t" + value.toString(); 105 } else { 106 str = value.toString(); 107 } 108 109 context.write(key, new Text(str)); 110 111 node.FormatNode(str); 112 113 // 没走到此节点 退出 114 if (node.getDistance().equals("inf")) 115 return; 116 117 // 重新计算源点A到各点的距离 118 for (int i = 0; i < node.getNodeNum(); i++) { 119 String k = node.getNodeKey(i); 120 String v = new String( 121 Integer.parseInt(node.getNodeValue(i)) + Integer.parseInt(node.getDistance()) + ""); 122 context.write(new Text(k), new Text(v)); 123 } 124 } 125 } 126 127 /** 128 * @author Edward 129 * 130 * B 10 (C,1) (D,2) 131 * B 8 => 132 * B 8 (C,1) (D,2) 133 * 134 */ 135 public static class ShortestPathReducer extends Reducer<Text, Text, Text, Text> { 136 137 protected void reduce(Text arg0, Iterable<Text> arg1, Context arg2) throws IOException, InterruptedException { 138 String min = null; 139 int i = 0; 140 String dis = "inf"; 141 Node node = new Node(); 142 for (Text t : arg1) { 143 i++; 144 dis = StringUtils.split(t.toString(), '\t')[0]; 145 146 // 如果存在inf节点，表示存在没有计算距离的节点。 147 // if(dis.equals("inf")) 148 // arg2.getCounter(eInf.COUNTER).increment(1L); 149 150 // 判断是否存在相邻节点，如果是则需要保留信息，并找到最小距离进行更新。 151 String[] strs = StringUtils.split(t.toString(), '\t'); 152 if (strs.length > 1) { 153 node.FormatNode(t.toString()); 154 } 155 156 // 第一条数据默认是最小距离 157 if (i == 1) { 158 min = dis; 159 } else { 160 if (dis.equals("inf")) 161 ; 162 else if (min.equals("inf")) 163 min = dis; 164 else if (Integer.parseInt(min) > Integer.parseInt(dis)) { 165 min = dis; 166 } 167 } 168 } 169 170 // 有新的最小值，说明还在进行优化计算，需要继续循环计算 171 if (!min.equals("inf")) { 172 if (node.getDistance().equals("inf")) 173 arg2.getCounter(eInf.COUNTER).increment(1L); 174 else { 175 if (Integer.parseInt(node.getDistance()) > Integer.parseInt(min)) 176 arg2.getCounter(eInf.COUNTER).increment(1L); 177 } 178 } 179 180 node.setDistance(min); 181 182 arg2.write(arg0, new Text(node.toString())); 183 } 184 } 185 }
Node.java
1 import org.apache.hadoop.util.StringUtils; 2 3 4 /** 5 * Created by Edward on 2016/7/15. 6 */ 7 public class Node { 8 private String distance; 9 private String[] adjs; 10 11 12 public String getDistance() { 13 return distance; 14 } 15 16 17 public void setDistance(String distance) { 18 this.distance = distance; 19 } 20 21 public String getKey(String str) 22 { 23 return str.substring(1, str.indexOf(",")); 24 } 25 26 public String getValue(String str) 27 { 28 return str.substring(str.indexOf(",")+1, str.indexOf(")")); 29 } 30 31 public String getNodeKey(int num) 32 { 33 return getKey(adjs[num]); 34 } 35 36 public String getNodeValue(int num) 37 { 38 return getValue(adjs[num]); 39 } 40 41 public int getNodeNum() 42 { 43 return adjs.length; 44 } 45 46 public void FormatNode(String str) 47 { 48 if(str.length() == 0) 49 return ; 50 51 String[] strs = StringUtils.split(str, '\t'); 52 53 adjs = new String[strs.length-1]; 54 for(int i=0; i<strs.length; i++) 55 { 56 if(i == 0) 57 { 58 setDistance(strs[i]); 59 continue; 60 } 61 this.adjs[i-1]=strs[i]; 62 } 63 } 64 65 public String toString() 66 { 67 String str = this.distance+"" ; 68 69 if(this.adjs == null) 70 return str; 71 72 for(String s:this.adjs) 73 { 74 str = str+"\t"+s; 75 } 76 return str; 77 } 78 79 public static void main(String[] args) 80 { 81 Node node = new Node(); 82 node.FormatNode("1 (A,20) (B,30)"); 83 System.out.println(node.distance+"|"+node.getNodeNum()+"|"+node.toString()); 84 } 85 }
