source: proiecte/HadoopA51/src/java/ro/pub/cs/pp/a51hadoop/search/Search.java @ 166

package ro.pub.cs.pp.a51hadoop.table;

import ro.pub.cs.pp.a51hadoop.config.Config;
import ro.pub.cs.pp.a51hadoop.common.*;
import ro.pub.cs.pp.a51hadoop.algorithm.*;
import ro.pub.cs.pp.a51hadoop.algorithm.testing.*;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

public class Search
{

        public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, ArrayWritable>
        {
                private HashReducer[] r;
                private Hashfn hashfn;
                private String needle;
                private ArrayList<String> needleGenerators;

                public void create_testing_problem(int nr_reducers, int id)
                {
                        hashfn = new DigitHashfn();
                        r = new DigitReducerGenerator().generate(nr_reducers, id);
                }

                public void create_a51_problem(int nr_reducers, int id)
                {
                        throw new RuntimeException("a51 not implemented");
                }

                public void configure(JobConf job)
                {
                        int nr_reducers = Config.N_R;
                        int k = job.getInt("mapred.mapper.table_id", 0);
                        int use_a51 = job.getInt("mapred.mapper.use_a51", 0);
                        this.needle = job.get("mapred.mapper.needle");
                        this.needleGenerators = new ArrayList<String>();

                        for(int i = 0; i < Config.N_R; i++)
                        {
                                String gen = applyLast(needle, i);
                                this.needleGenerators.add(gen);
                        }


                        if (use_a51 != 0)
                                create_a51_problem(nr_reducers, k);
                        else
                                create_testing_problem(nr_reducers, k);
                }

                public String applyAll(String txt)
                {
                        for (int i = 0; i < Config.N_R; i++)
                        {
                                txt = r[i].apply(hashfn.hashfn(txt));
                        }
                        return txt;
                }

                public String applyLast(String hashstr, int n)
                {
                        for (int i = Config.N_R - n; i < Config.N_R - 1; i++)
                        {
                                hashstr = hashfn.hashfn(r[i].apply(hashstr));
                        }
                        return r[r.length - 1].apply(hashstr);
                }


                public String applyFirst(String txt, int n)
                {
                        for (int i = 0; i < n; i++)
                        {
                                txt = hashfn.hashfn(r[i].apply(txt));
                        }
                        return txt;
                }


                public void map(LongWritable key, Text value, OutputCollector<Text, ArrayWritable> output, Reporter reporter) throws IOException
                {

                        String line = value.toString();
                        StringTokenizer tokenizer = new StringTokenizer(line);
                        if (!tokenizer.hasMoreTokens())
                                return;
                        String end = tokenizer.nextToken();
                        List<Text> list = new ArrayList<Text>();
                        ArrayWritable arr = new TextArrayWritable();
                        Text word = new Text();
                        word.set(needle);
                        while (tokenizer.hasMoreTokens())
                        {
                                String start = tokenizer.nextToken();
                                for (int i = 0; i < needleGenerators.size(); i++)
                                {
                                        if (start.equals(needleGenerators.get(i)))
                                        {
                                                String hashStr = applyFirst(start, Config.N_R - i);
                                                if (!hashStr.equals(needle))
                                                        continue;
                                                Text t = new Text();
                                                t.set(start);
                                                list.add(t);
                                        }
                                }
                        }

                        arr.set(list.toArray(new Text[0]));
                        output.collect(word, arr);
                }
        }

        public static class Reduce extends MapReduceBase implements Reducer<Text, ArrayWritable, Text, ArrayWritable>
        {

                public void reduce(Text key, Iterator<ArrayWritable> values, OutputCollector<Text, ArrayWritable> output, Reporter reporter) throws IOException
                {
                        List<Text> list = new ArrayList<Text>();
                        while (values.hasNext())
                        {
                                //Text[] arr = (Text[]) ;
                                Writable[] arr = values.next().get();
                                for (int i = 0; i < arr.length; i++)
                                {
                                        Text t = (Text) arr[i];
                                        list.add(t);
                                }
                        }
                        ArrayWritable arr = new TextArrayWritable();
                        arr.set(list.toArray(new Text[0]));
                        output.collect(key, arr);
                }
        }

        public static void run(String input, String output, int k) throws Exception
        {
                JobConf conf = new JobConf(TableGen.class);
                String kstr = "" + k;
                conf.setJobName("tablegen");
                conf.setOutputKeyClass(Text.class);
                conf.setOutputValueClass(TextArrayWritable.class);

                conf.setMapperClass(Map.class);
                conf.setCombinerClass(Reduce.class);
                conf.setReducerClass(Reduce.class);

                conf.setInputFormat(TextInputFormat.class);
                conf.setOutputFormat(TextOutputFormat.class);

                FileInputFormat.setInputPaths(conf, new Path(input));
                FileOutputFormat.setOutputPath(conf, new Path(new Path(output), kstr));

                JobClient.runJob(conf);
        }

        public static void main(String[] args) throws Exception
        {
                for(int i = 0; i < Config.N_TABLES; i++)
                {
                        run(args[0], args[1], i);
                        System.out.println("LAG:XXXX:XXXX: i=" + i);
                }
        }
}