package pl.edu.icm.coansys.commons.hbase;

import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 *
 * @author akawa
 */
public class SequenceFileKeysSampler implements Tool {

    private Configuration conf;
    private final String OUTPUT_KEYS_FILE_NAME_KEY = "output.keys.file.name";
    private final String OUTPUT_KEYS_FILE_NAME_DEFAULT_VALUE = "keys";
    private final String SAMPLER_FREQUENCY_KEY = "sampler.frequency";
    private final float SAMPLER_FREQUENCY_DEFAULT_VALUE = 0.1f;
    private final String SAMPLER_NUM_SAMPLES_KEY = "sampler.num.samples";
    private final int SAMPLER_NUM_SAMPLES_DEFAULT_VALUE = 100;

    @Override
    public void setConf(Configuration conf) {
        this.conf = conf;
    }

    @Override
    public Configuration getConf() {
        return conf;
    }

    @Override
    public int run(String[] args) throws Exception {
        String sequenceFileInput = args[0];
        int regionCnt = Integer.parseInt(args[1]);
        String keysDirOutput = args[2];
        int result = createParitionFile(sequenceFileInput, regionCnt, keysDirOutput);
        return result;
    }

    private int createParitionFile(String sequenceFileInput, int regionCnt, String keysDirOutput) throws IOException, ClassNotFoundException, InterruptedException {
        Path input = new Path(sequenceFileInput);
        Job sampler = new Job(getConf());

        sampler.setNumReduceTasks(regionCnt);
        sampler.setInputFormatClass(SequenceFileInputFormat.class);
        sampler.setOutputFormatClass(SequenceFileOutputFormat.class);
        sampler.setOutputKeyClass(BytesWritable.class);
        SequenceFileInputFormat.addInputPath(sampler, input);

        Configuration config = sampler.getConfiguration();
        
     
        InputSampler.Sampler<BytesWritable, NullWritable> inputSampler =
                new InputSampler.RandomSampler<BytesWritable, NullWritable>(
                config.getFloat(SAMPLER_FREQUENCY_KEY, SAMPLER_FREQUENCY_DEFAULT_VALUE),
                config.getInt(SAMPLER_NUM_SAMPLES_KEY, SAMPLER_NUM_SAMPLES_DEFAULT_VALUE));

        Path partitionDir = new Path(keysDirOutput);
        partitionDir = partitionDir.makeQualified(partitionDir.getFileSystem(config));
        Path partitionFile = new Path(partitionDir, config.get(OUTPUT_KEYS_FILE_NAME_KEY, OUTPUT_KEYS_FILE_NAME_DEFAULT_VALUE));
        TotalOrderPartitioner.setPartitionFile(config, partitionFile);
        InputSampler.<BytesWritable, NullWritable>writePartitionFile(sampler, inputSampler);

        return 0;
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 3) {
            usage("Wrong number of arguments: " + otherArgs.length);
            System.exit(-1);
        }

        int result = ToolRunner.run(conf, new SequenceFileKeysSampler(), otherArgs);
        System.exit(result);
    }

    private static void usage(String info) {
        System.out.println(info);
        System.out.println("Three parameters needed: <sequence-file-input, number-of-regions, keys-dir-output>");
        System.out.println("Example: hadoop jar target/commons-1.0-SNAPSHOT.jar " + SequenceFileKeysSampler.class.getName() + " bazekon-20120228.sf 20 sf-split");
    }
}



// hadoop jar target/commons-1.0-SNAPSHOT.jar pl.edu.icm.coansys.commons.hbase.SequenceFileKeysSampler bazekon-20120228.sf 20 sf-split
// export HBASE_CLASSPATH=target/commons-1.0-SNAPSHOT.jar
// hbase org.apache.hadoop.hbase.util.RegionSplitter -D split.algorithm=pl.edu.icm.coansys.commons.hbase.SequenceFileSplitAlgorithm -c 2 -f c:m test_table1
@@ -549,60 +549,8 @@
 er;%0A
-import org.apache.hadoop.util.GenericOptionsParser;%0A
 impo
@@ -1158,16 +1158,189 @@
 UE = 100
+0;%0A    %0A    private static final String%5B%5D DEFAULT_ARGS = %7B%0A        %22/home/akawa/Documents/git-projects/CoAnSys/importers/src/main/bash/noncompressed.sf%22,%0A        %2220%22,%0A    %7D
 ;%0A%0A    @
@@ -1564,32 +1564,176 @@
 rows Exception %7B
+%0A%0A        if (args.length %3C 2) %7B%0A            usage(%22Wrong number of arguments: %22 + args.length);%0A            System.exit(-1);%0A        %7D%0A        
 %0A        String 
@@ -1816,48 +1816,8 @@
 %5D);%0A
-        String keysDirOutput = args%5B2%5D;%0A
     
@@ -1884,23 +1884,8 @@
 nCnt
-, keysDirOutput
 );%0A 
@@ -1991,30 +1991,8 @@
 nCnt
-, String keysDirOutput
 ) th
@@ -2511,30 +2511,16 @@
 tion();%0A
-        %0A     
 %0A       
@@ -2848,148 +2848,8 @@
 );%0A%0A
-        Path partitionDir = new Path(keysDirOutput);%0A        partitionDir = partitionDir.makeQualified(partitionDir.getFileSystem(config));%0A
     
@@ -2886,22 +2886,8 @@
 ath(
-partitionDir, 
 conf
@@ -3123,17 +3123,16 @@
 mpler);%0A
-%0A
         
@@ -3214,289 +3214,8 @@
 n %7B%0A
-        Configuration conf = new Configuration();%0A        String%5B%5D otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();%0A        if (otherArgs.length %3C 3) %7B%0A            usage(%22Wrong number of arguments: %22 + otherArgs.length);%0A            System.exit(-1);%0A        %7D%0A%0A
     
@@ -3250,14 +3250,8 @@
 run(
-conf, 
 new 
@@ -3281,14 +3281,9 @@
 (), 
-otherA
+a
 rgs)
@@ -3432,12 +3432,10 @@
 n(%22T
-hree
+wo
  par
@@ -3493,25 +3493,8 @@
 ions
-, keys-dir-output
 %3E%22);
@@ -3644,25 +3644,16 @@
 28.sf 20
- sf-split
 %22);%0A    
@@ -3659,356 +3659,4 @@
  %7D%0A%7D
-%0A%0A%0A%0A// hadoop jar target/commons-1.0-SNAPSHOT.jar pl.edu.icm.coansys.commons.hbase.SequenceFileKeysSampler bazekon-20120228.sf 20 sf-split%0A// export HBASE_CLASSPATH=target/commons-1.0-SNAPSHOT.jar%0A// hbase org.apache.hadoop.hbase.util.RegionSplitter -D split.algorithm=pl.edu.icm.coansys.commons.hbase.SequenceFileSplitAlgorithm -c 2 -f c:m test_table1

refactor with GenericOptionParser