package com.xirideguang.resys;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.hadoop.compression.lzo.DistributedLzoIndexer;
import com.hadoop.mapreduce.LzoTextInputFormat;

/**
 * Job名称： ReverseIdJob
 * 
 * 输入:"itme_a_id" + "\t" + "item_b_id" + "\t" + "value" + "\t" + "timestamp"
 * 
 * 输出:"itme_b_id" + "\t" + "item_a_id" + "\t" + "value" + "\t" + "timestamp"
 * 
 * Job实现功能： 翻转数据中的id
 * 
 * 依据原理：翻转操作id和被 操作id。便于计算使用。
 */

public class ReverseIdJob extends Configured implements Tool {
    private static String TAB = "\t";

    public static class ReverseIdMapper extends
	    Mapper<LongWritable, Text, Text, Text> {
	@Override
	protected void map(LongWritable key, Text value, Context context)
		throws IOException, InterruptedException {
	    Text outputKey = new Text();
	    Text outputValue = new Text();

	    StringTokenizer valStringTokenizer = new StringTokenizer(
		    value.toString(), TAB);
	    String keyString = valStringTokenizer.nextToken();
	    keyString = valStringTokenizer.nextToken() + TAB + keyString;
	    String valString = valStringTokenizer.nextToken() + TAB
		    + valStringTokenizer.nextToken();// 翻转itemid

	    outputKey.set(keyString);
	    outputValue.set(valString);
	    context.write(outputKey, outputValue);

	}
    }

    public static void main(String[] args) throws Exception {
	int res = ToolRunner.run(new Configuration(), new ReverseIdJob(), args);
	System.exit(res);
    }

    public int run(String[] args) throws Exception {

	Job job = new Job(getConf(), "ResysTest ReverseIdJob");

	job.setJarByClass(ReverseIdJob.class);

	Configuration conf = job.getConfiguration();

	conf.set("io.compression.codecs",
		"org.apache.hadoop.io.compress.DefaultCodec,"
			+ "org.apache.hadoop.io.compress.GzipCodec,"
			+ "org.apache.hadoop.io.compress.BZip2Codec,"
			+ "org.apache.hadoop.io.compress.SnappyCodec,"
			+ "com.hadoop.compression.lzo.LzopCodec,"
			+ "com.hadoop.compression.lzo.LzoCodec");
	String ifOutputCompress = conf.get("if_output_compress");
	if (!"false".equals(ifOutputCompress)) {
	    conf.set("mapred.output.compress", "true");
	    String codec = conf.get("codec");
	    if (codec != null) {
		conf.set("mapred.output.compression.codec", codec);
	    } else {
		conf.set("mapred.output.compression.codec",
			"com.hadoop.compression.lzo.LzopCodec");
	    }
	}
	String ifMapCompress = conf.get("if_map_compress");
	if (!"false".equals(ifMapCompress)) {
	    conf.set("mapred.compress.map.output", "true");
	    conf.set("mapred.map.output.compression.codec",
		    "com.hadoop.compression.lzo.LzoCodec");
	}
	conf.set("mapred.task.timeout", "0");

	String inputPath = conf.get("input");
	String outputPath = conf.get("output");
	int NumReduceTasks = Integer.parseInt(conf.get("num_reduce_tasks"));

	String ifIndexed = conf.get("if_indexed");
	if (!"false".equals(ifIndexed)) {
	    job.setInputFormatClass(LzoTextInputFormat.class);
	}
	FileInputFormat.addInputPath(job, new Path(inputPath));
	FileOutputFormat.setOutputPath(job, new Path(outputPath));
	job.setNumReduceTasks(NumReduceTasks);

	job.setMapperClass(ReverseIdMapper.class);

	job.setMapOutputKeyClass(Text.class);
	job.setMapOutputValueClass(Text.class);

	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(Text.class);

	int result = job.waitForCompletion(true) ? 0 : 1;
	if (!"false".equals(ifOutputCompress)) {
	    String ifIndex = conf.get("if_index");
	    if (!"false".equals(ifIndex)) {
		System.gc();
		DistributedLzoIndexer lzoIndexer = new DistributedLzoIndexer();
		Configuration indexConf = new Configuration();
		indexConf.set("io.compression.codecs",
			"com.hadoop.compression.lzo.LzopCodec");
		lzoIndexer.setConf(indexConf);
		lzoIndexer.run(new String[] { outputPath });
	    }
	}
	System.exit(result);
	return 0;
    }
}
