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

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.sql.*;
import java.util.Properties;



public class Transformer {

	public static class TransformerMapper extends Mapper<Text, Text, Text, MapWritable>
    {

        public void map(Text key, Text value, Context context) throws IOException, InterruptedException
        {
        	MapWritable mm = new MapWritable();

        	StringTokenizer itr = new StringTokenizer(value.toString(),"\t");

        	int i = 0;

            while (itr.hasMoreTokens())
            {
            	String t = itr.nextToken().toString() + "-map-";
            	mm.put(new Text("artistName"+i), new Text(t));
            	i++;
            }
            context.write(key, mm);
        }
    }
    public static class TransformerReducer extends Reducer<Text,MapWritable,NullWritable,Text>
    {
    	private NullWritable nullKey = NullWritable.get();
    	private Text outputText = new Text();

        public void reduce(Text key, Iterable<MapWritable> values, Context context) throws IOException, InterruptedException
        {
            String outputTxt = "";
            int i = 0;
            while(values.iterator().hasNext()){
            	MapWritable entry = values.iterator().next();
            	outputTxt += entry.get(new Text("artistName"+i++)).toString()+key.toString();
            }
            outputText.set(outputTxt);
            context.write(nullKey, outputText);
        }
    }
	public static void main(String[] args) throws Exception{
		// TODO Auto-generated method stub
		Configuration conf = new Configuration();
		conf.set("key.value.separator.in.input.line","|");
        Job job = new Job(conf, "transformer");
        job.setJarByClass(Transformer.class);

        job.setMapperClass(TransformerMapper.class);
        job.setReducerClass(TransformerReducer.class);

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

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


        job.setInputFormatClass(KeyValueTextInputFormat.class);


        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}