现阶段MapReduce框架 实现简单图的算法

刚开始接触hadoop的mapreduce并行计算的编程框架，使用的是java语言，对于一些简单的日志文档处理，相当的容易上手，但是经过一段时间的学习调研，发现用其实现一些图的算法，相当蹩脚，效率很低。。。

 

下面我列出下mapreduce实现图的单源最短路径的算法（伪代码）这里假设的是每个节点之间是单源节点

1: class Mapper
2: method Map(nid n,node N)
3: d ! N.Distance
4: Emit(nid n,N) Passalonggraphstructure
5: forall nodeid m ! N.AdjacencyListdo
6: Emit(nid m,d +1) Emitdistancestoreachablenodes

1: class Reducer
2:       method Reduce(nid m,[d ,d ,...])
3:      dmin=Max
4:      M=0
5:      forall d ! counts [d ,d ,...] do
6:     if IsNode(d) then
7:     M=d 
8:    elseif d<dmin then 
       dmin=d
9:    M.Distance=dmin
10:  Emit(nid m,node M)

 其中的要点是（采用的广度优先搜索的思想）

 

这只是其中的一次迭代，每次迭代都是一个job，而且每个job之间需要传输一个特殊的键值对  包含整个图的完整信息。

还有一个问题整体的迭代次数，什么时候停止呢  如何判断，我这里采用的Counter计数器，记录到达源节点的距离为Max的个数，根据这个个数来判断是否该停止工作，工作之间的数据传输，最好采用二进制文件来传输方便处理。图在mapreduce中存储是采用临界表的方式进行存储，比较方便

 

如果实现节点之间的边带权值的也很简单  只需要将上述的伪代码  d+1 改为d+对应的全知即可

 

整个工作需要一个驱动程序来控制作业。。下面是一个简单的实现 伪代码

public static void main(String[] args) throws Exception
	{
		long count = 0;
		int it=1;
		String input="/Graph/input/";
		String output="/Graph/output1";
		do
		{
			Job job = getNodeJob(input, output);
			job.waitForCompletion(true);
			Counters counter = job.getCounters();
			count = counter.findCounter(ReachableNodes.Reduce).getValue();
			
			input="/Graph/output"+it+"/part-r-00000";
			it++;
			output="/Graph/output"+it;
		} while (count != agr[0]);//所有节点都可达的时候结束迭代--
	}
	
	public static Job getNodeJob(String inPath,String outPath) throws Exception
	{
		Configuration conf = new Configuration();
		Job job=new Job(conf,"Iterable Node");
		
		job.setJarByClass(GraphDrive.class);
		
		job.setNumReduceTasks(1);

		job.setMapperClass(GraphMapper.class);
		job.setReducerClass(GraphReducer.class);
		
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(BFSNode.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(BFSNode.class);
		
		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
		
		FileInputFormat.addInputPath(job, new Path(inPath));
		FileOutputFormat.setOutputPath(job, new Path(outPath));
		
		FileSystem.get(job.getConfiguration()).delete(new Path(outPath), true);//如果文件已存在删除
		
		return job;	
	}

 

 

下面我来说说我的问题，能帮我解惑的朋友谢谢指正，当把一个图存放到一个文件的时候 也就是一个map的时候 ，比较容易实现--很容易收敛，，但是如果采用多个map任务的时候  收敛会很慢  甚至死循环。。。很困惑