java – Hadoop Map Reduce For Google web graph

我们已经作为一项任务给出了创建map reduce函数的任务,该函数将为google web graph列表中的每个节点n输出您可以从3跳中的节点n开始的节点. (实际数据可在此处找到： http://snap.stanford.edu/data/web-Google.html)
以下是列表中项目的示例：

1 2 
1 3 
2 4 
3 4 
3 5 
4 1 
4 5 
4 6 
5 6

从上面的示例图将是这样的

在上面的简化示例中,例如节点1的路径是
α[1 – > 2 – > 4 – > 1],[1 – > 2 – > 4 – > 5],[1 – > 2 – > 4 – > 6],[1 – > 3 – > 4 – > 1],
[1 – > 3 – > 4 – > 5],[1 – > 3 – > 4 – > 6]και[1 – > 3 – > 5 – > 6]
因此地图缩小将为节点1输出顶点1,5,6(
(a)每个顶点只能计算一次,并且
(b)我们仅在存在长度为3的圆形路径时包括当前顶点,如示例[1 – > 2 – > 4 – > 1]和[1 – > 3 – > 4 – > 1].

我很失落,因为我相信这需要图论和算法的知识,而且我们没有被教过任何与之相关的知识.

如果有人能给我正确的方向开始,我将不胜感激. (我已经研究过最短路径理论,但我不确定它是否会对这个特定的练习有用)

在此先感谢,并有一个愉快的假期.

编辑

我尝试创建adjucency列表,但我希望输出为“vertexID”“node1 node2 node3 node4 …”我看到在输出文件中我的reducer将每个顶点Id的列表拆分为三对.

例如,如果我有一个连接到Z,X,C,V,B,N,M,G,H,J,K,L的顶点A我将它输出为

Z,C

A V,N

A M,H

A J,L

下面是我的mapper和reducer

public class AdjacentsListDriver extends Configured implements Tool {

    @Override
    public int run(String[] args) throws Exception {



        Configuration conf = getConf();
        Job job = Job.getInstance(conf);
        job.setJobName("Test driver");
        job.setJarByClass(AdjacentsListDriver.class);

        String[] arg0 = new GenericoptionsParser(conf,args).getRemainingArgs();
        if (arg0.length != 2) {
            System.err.println("Usage: hadoop jar <my_jar> <input_dir> <output_dir>");
            System.exit(1);
        }

        Path in = new Path(arg0[0]);
        Path out = new Path(arg0[1]);

        FileInputFormat.setInputPaths(job,in);
        FileOutputFormat.setoutputPath(job,out);

        job.setMapperClass(ListMapper.class);
        job.setReducerClass(ListReducer.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setoutputFormatClass(textoutputFormat.class);

        job.setoutputKeyClass(Text.class);
        job.setoutputValueClass(Text.class);    
        job.waitForCompletion(true);

        return 0;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(),new AdjacentsListDriver(),args);
        System.exit(res);

    }



}





/**
 * @author George
 * Theoretically this takes a key(vertexID) and maps all nodes that are connected to it in one hop....
 *
 */
public class ListMapper extends Mapper<LongWritable,Text,Text> {

    private Text vertexID = new Text();
    //private LongWritable vertice= new LongWritable(0);
    private Text vertice=new Text();

    public void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException {

        String line = value.toString();
        StringTokenizer itr = new StringTokenizer(line,"\n");
        StringTokenizer itrInside;

        //vertice=new LongWritable(Long.valueOf(value.toString()).longValue());


        while (itr.hasMoretokens()) {
            if(itr.countTokens() > 2){

            }//ignore first line ??
            else{
                itrInside=new StringTokenizer(itr.toString());
                vertexID.set(itr.nextToken());

                while(itrInside.hasMoretokens()){               
                    vertice.set(itrInside.nextToken());
                    context.write(vertexID,value);
                }           
            }
        }

    }

}

@override
public class ListReducer extends Reducer<Text,Text>{
    public void reduce(Text key,Iterable<Text> values,InterruptedException {

        String vertices="";

        for (Text value : values) {
            if(vertices=="")
                vertices+=value.toString();
            else
                vertices=vertices+","+value.toString();         
        }

        Text value=new Text();
        value.set(vertices);
        context.write(key,value);

    }

}

map input tuple (X,Y): 
   emit (X,Y)

reduce input (X,Y[]) :
  emit (X,Y[])

HopMessage: Origin (VertexID) | HopCounter(Integer)

map input (VertexID key,either HopMessage or List<VertexID> adjacents):
  if(iterationNumber == 1): // only in the first iteration to kick off
     for neighbour in adjacents:
        emit (neighbour,new HopMessage(key,0))
  emit (key,adjacents or HopMessage) // for joining in the reducer

1 2 // graph 
2 1 // hop message
2 3 // graph
3 1 // hop message
3 4 // graph
4 1 // hop message
4 - // graph

reducer input(VertexID key,List<either HopMessage or List<VertexID> neighbours> values):
 for hopMessage in values:
    hopMessage.counter += 1
    if (hopMessage.counter == 3) 
       emit to some external resource (HopMessage.origin,key)
    else 
       for neighbour in neighbours of key:
          emit (neighbour,hopMessage)
    emit (key,neighbours)

public static class NthHopVertex extends Vertex<Text,NullWritable,HopMessage> {

    @Override
    public void compute(Iterable<HopMessage> messages) throws IOException {
      if (getSuperstepCount() == 0L) {
        HopMessage msg = new HopMessage();
        msg.origin = getVertexID().toString();
        msg.hopCounter = 0;
        sendMessagetoNeighbors(msg);
      } else {
        for (HopMessage message : messages) {
          message.hopCounter += 1;
          if (message.hopCounter == 3) {
            getPeer().write(new Text(message.origin),getVertexID());
            VoteToHalt();
          } else {
            sendMessagetoNeighbors(message);
          }

        }
      }
    }
  }

1=[1,6]
2=[2,3,6]
3=[2,6]
4=[4,5]