package org.myorg;

	import java.io.*;
	import java.util.*;
	import org.apache.hadoop.fs.*;
	import org.apache.hadoop.conf.*;
	import org.apache.hadoop.io.*;
	import org.apache.hadoop.mapred.*;
	import org.apache.hadoop.util.*;
	import org.apache.log4j.Logger;
	import org.apache.log4j.LogManager;
	import org.apache.log4j.xml.DOMConfigurator;

	public class ParallelIndexation {
		//public static native long Traveser(String Path);

		//public static native void Configure(String Path);

		//static {
		//	System.loadLibrary("nativelib");
		//}
		public static class Map extends MapReduceBase implements
				Mapper<LongWritable, Text, Text, LongWritable> {
			private final static LongWritable zero = new LongWritable(0);
			private Text word = new Text();
			private static final Logger logger = LogManager.getLogger(Map.class.getName());
			//DOMConfigurator.configure("/export/hadoop-1.0.1/log4j.xml");

			public void map(LongWritable key, Text value,
					OutputCollector<Text, LongWritable> output, Reporter reporter)
					throws IOException {
				DOMConfigurator.configure("/folder/log4j.xml");
				Configuration conf = new Configuration();
				FileSystem fs = FileSystem.get(conf);
				Path localPath = new Path("/export/hadoop-1.0.1/bin/input/paths.txt");
				Path hdfsPath=new Path("hdfs://192.168.1.8:7000/user/hadoop/paths.txt");
				Path localPath1 = new Path("/usr/countcomputers.txt");	              
				Path hdfsPath1=new Path("hdfs://192.168.1.8:7000/user/hadoop/countcomputers.txt");
				if (!fs.exists(hdfsPath))
				{
					fs.copyFromLocalFile(localPath, hdfsPath);
				};
				if (!fs.exists(hdfsPath1))
				{
					fs.copyFromLocalFile(localPath1, hdfsPath1);
				};			
				FSDataInputStream in = fs.open(hdfsPath);
				BufferedReader br = new BufferedReader(new InputStreamReader(in));
				String line = br.readLine();
				// String line = value.toString();
				BufferedReader br1=new BufferedReader(new InputStreamReader(fs.open(hdfsPath1)));
				int CountComputers;
				/* FileInputStream fstream = new FileInputStream(
						"/usr/countcomputers.txt");
				DataInputStream input = new DataInputStream(fstream);
				BufferedReader br = new BufferedReader(new InputStreamReader(input)); */
				String result=br1.readLine();
				CountComputers=Integer.parseInt(result);
				// in.close();
				// fstream.close();
				ArrayList<String> paths = new ArrayList<String>();
				StringTokenizer tokenizer = new StringTokenizer(line, "|");
				while (tokenizer.hasMoreTokens()) {
					paths.add(tokenizer.nextToken());
				}
				for (int i=0; i<paths.size(); i++)
					{
						logger.debug("paths[i]=" + paths.get(i) + "\n");	
					}
				logger.debug("CountComputers=" + CountComputers + "\n");
				String[] ConcatPaths = new String[CountComputers];
				int NumberOfElementConcatPaths = 0;
				if (paths.size() % CountComputers == 0) {
					for (int i = 0; i < CountComputers; i++) {
						ConcatPaths[i] = paths.get(NumberOfElementConcatPaths);
						NumberOfElementConcatPaths += paths.size() / CountComputers;
						for (int j = 1; j < paths.size() / CountComputers; j++) {
							ConcatPaths[i] += "\n"
									+ paths.get(i * paths.size() / CountComputers
											+ j);
						}
					}
				} else {
					NumberOfElementConcatPaths = 0;
					for (int i = 0; i < paths.size() % CountComputers; i++) {
						ConcatPaths[i] = paths.get(NumberOfElementConcatPaths);
						NumberOfElementConcatPaths += paths.size() / CountComputers
								+ 1;
						for (int j = 1; j < paths.size() / CountComputers + 1; j++) {
							ConcatPaths[i] += "\n"
									+ paths.get(i
											* (paths.size() / CountComputers + 1)
											+ j);
						}
					}
					for (int k = paths.size() % CountComputers; k < CountComputers; k++) {
						ConcatPaths[k] = paths.get(NumberOfElementConcatPaths);
						NumberOfElementConcatPaths += paths.size() / CountComputers;
						for (int j = 1; j < paths.size() / CountComputers; j++) {
							ConcatPaths[k] += "\n"
									+ paths.get((k - paths.size() % CountComputers)
											* paths.size() / CountComputers
											+ paths.size() % CountComputers
											* (paths.size() / CountComputers + 1)
											+ j);
						}
					}
				}
				for (int i = 0; i < ConcatPaths.length; i++) {
					word.set(ConcatPaths[i]);
					output.collect(word, zero);
				}
				in.close();
			}
		}