public void reduce(Text key,Iterator<ObjectWritable> values,OutputCollector<Text,Route> output,Reporter reporter) throws IOException {
  String url=key.toString();
  Node node=null;
  List<LinkDatum> outlinkList=new ArrayList<LinkDatum>();
  while (values.hasNext()) {
    ObjectWritable objWrite=values.next();
    Object obj=objWrite.get();
    if (obj instanceof LinkDatum) {
      outlinkList.add((LinkDatum)obj);
    }
 else     if (obj instanceof Node) {
      node=(Node)obj;
    }
  }
  if (node != null) {
    int numInlinks=node.getNumInlinks();
    if (numInlinks > 0) {
      for (      LinkDatum datum : outlinkList) {
        String outlinkUrl=datum.getUrl();
        Route route=new Route();
        route.setFound(false);
        route.setLookingFor(url);
        route.setOutlinkUrl(outlinkUrl);
        output.collect(new Text(outlinkUrl),route);
      }
    }
  }
}

public void reduce ( Text key , Iterator < ObjectWritable > values , OutputCollector < Text , Route > output , Reporter reporter ) throws IOException { String url = key . toString ( ) ; Node node = null ; List < LinkDatum > outlinkList = new ArrayList < LinkDatum > ( ) ; while ( values . hasNext ( ) ) { ObjectWritable objWrite = values . next ( ) ; Object obj = objWrite . get ( ) ; if ( obj instanceof LinkDatum ) { outlinkList . add ( ( LinkDatum ) obj ) ; } else if ( obj instanceof Node ) { node = ( Node ) obj ; } } if ( node != null ) { int numInlinks = node . getNumInlinks ( ) ; if ( numInlinks > 0 ) { for ( LinkDatum datum : outlinkList ) { String outlinkUrl = datum . getUrl ( ) ; Route route = new Route ( ) ; route . setFound ( false ) ; route . setLookingFor ( url ) ; route . setOutlinkUrl ( outlinkUrl ) ; output . collect ( new Text ( outlinkUrl ) , route ) ; } } } }
takes any node that has inlinks and sets up a route for all of its outlinks . these routes will then be followed to a maximum depth inside of the looper job .
