5 Clicks to Jesus demo

VerifyLinksJob.java

package edu.ncsu.sdc.wikisearch.jobs;

import java.io.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;

import edu.ncsu.sdc.wikisearch.tables.*;

/**
 * MapReduce classes for VerifyIndex Tool
 */
public class VerifyLinksJob {
    public static class Map extends Mapper<IntWritable, NullWritable, IntWritable, IntWritable> {
        private final String seed = "Jesus";
        private ArticleIdentifierTable idLookupTable;
        private LinkTable linkTable;

        protected void setup(Context context) throws IOException, InterruptedException {
            Configuration conf = context.getConfiguration();
            
            this.idLookupTable = ArticleIdentifierTable.openDefaultTable(conf);
            this.linkTable = LinkTable.openDefaultTable(conf);
        }

        public void map(IntWritable key, NullWritable value, Context context) throws IOException, InterruptedException {
            int[] distances = new int[this.idLookupTable.getNumArticles()];
            int r = 0;
            
            for(int i = 0; i < distances.length; i++) {
                distances[i] = -1;
            }
            
            distances[this.idLookupTable.getArticleIdentifierByName(this.seed)] = 0;

            while(r < 5) {
                for(int i = 0; i < distances.length; i++) {
                    if(distances[i] == r) {
                        for(ArticleNeighbor neighbor : this.linkTable.getNeighbors(i)) {
                            int j = neighbor.getNeighbor();

                            if(neighbor.getInCount() > 0 && distances[j] == -1) {
                                distances[j] = r + 1;
                            }
                        }
                    }
                }

                r++;
            }
            
            for(int i = 0; i < distances.length; i++) {
                if(distances[i] >= 0) {
                    context.write(new IntWritable(i), new IntWritable(distances[i]));
                } else {
                    int outLinkCount = 1;
                    for(ArticleNeighbor neighbor : this.linkTable.getNeighbors(i)) {
                        outLinkCount += neighbor.getOutCount();
                    }
                    context.write(new IntWritable(i), new IntWritable(-outLinkCount));
                }
            }
        }
    }

    public static class Reduce extends Reducer<IntWritable, IntWritable, Text, IntWritable> {
        private ArticleIdentifierTable idLookupTable;
        
        protected void setup(Context context) throws IOException, InterruptedException {
            this.idLookupTable = ArticleIdentifierTable.openDefaultTable(context.getConfiguration());
        }

        public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            Text articleName = new Text(this.idLookupTable.getArticleNameByIdentifier(key.get()));
            
            for(IntWritable v : values) {
                context.write(articleName, v);
            }
        }
    }
}