source: proiecte/HadoopA51/src/java/ro/pub/cs/pp/a51hadoop/table/TableGen.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 TableGen
{

        public static class R
        {
                private int i;
                private int k;
                public R(int i, int k)
                {
                        this.i = i;
                        this.k = k;
                }

                public String apply(String strX)
                {
                        try {
                                int x = Integer.parseInt(strX);
                                int last_digit = 0;
                                switch (i)
                                {
                                case 0:
                                        last_digit = i % 3;
                                        break;
                                case 1:
                                        last_digit = i % 3 + 3;
                                        break;
                                case 2:
                                        last_digit = i % 3 + 6;
                                        break;
                                default:
                                        last_digit = 9;
                                        break;
                                }
                                x = x / 10 * 10 + last_digit;

                                x = x ^ k;

                                String ret = "" + x;
                                while(ret.length() < strX.length())
                                        ret = "0" + ret;
                                if (ret.length() > strX.length())
                                        ret = ret.substring(0, strX.length());
                                return ret;
                        } catch (Exception e) {
                                return strX;
                        }
                }
        }

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

                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);
                        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 < txt.length(); i++)
                        {
                                txt = r[i].apply(hashfn.hashfn(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);
                        while (tokenizer.hasMoreTokens())
                        {
                                String wordStr = tokenizer.nextToken();
                                String hashStr = applyAll(wordStr);
                                Text word = new Text();
                                Text hash = new Text();
                                List<Text> list = new ArrayList<Text>();

                                word.set(wordStr);
                                hash.set(hashStr);
                                list.add(hash);
                                ArrayWritable arr = new TextArrayWritable();
                                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);
                }
        }
}