<commit_before>package com.sensei.indexing.hadoop.job;

import java.io.IOException;
import java.text.NumberFormat;
import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.log4j.Logger;

import com.sensei.indexing.hadoop.keyvalueformat.IntermediateForm;
import com.sensei.indexing.hadoop.keyvalueformat.Shard;
import com.sensei.indexing.hadoop.map.SenseiMapper;
import com.sensei.indexing.hadoop.reduce.FileSystemDirectory;
import com.sensei.indexing.hadoop.reduce.IndexUpdateOutputFormat;
import com.sensei.indexing.hadoop.reduce.SenseiCombiner;
import com.sensei.indexing.hadoop.reduce.SenseiReducer;
import com.sensei.indexing.hadoop.util.LuceneUtil;
import com.sensei.indexing.hadoop.util.MRJobConfig;
import com.sensei.indexing.hadoop.util.SenseiJobConfig;

public class MapReduceJob extends Configured {

	private static final NumberFormat NUMBER_FORMAT = NumberFormat.getInstance();
	private static final Logger logger = Logger.getLogger(MapReduceJob.class);
	
	  public JobConf createJob(Class MRClass) throws IOException {
		    
		    Configuration conf = getConf();
		    Path[] inputPaths;
		    Path outputPath;
		    Shard[] shards = null;
			int numMapTasks = conf.getInt("mapreduce.job.maps", 2);
			int numShards = conf.getInt(SenseiJobConfig.NUM_SHARDS, 2);
//			inputPaths = FileInputFormat.getInputPaths(jobConf);
			
		    String dirs = conf.get("mapreduce.input.fileinputformat.inputdir", null);
		    logger.info("dirs:"+ dirs);
		    String [] list = StringUtils.split(dirs);
		    logger.info("length after split:"+ list.length);
		    inputPaths = new Path[list.length];
		    for (int i = 0; i < list.length; i++) {
		    	inputPaths[i] = new Path(StringUtils.unEscapeString(list[i]));
		    }
		    logger.info("path[0] is:" + inputPaths[0]);
		    	    
			outputPath = new Path(conf.get("mapreduce.output.fileoutputformat.outputdir"));
			String indexPath = conf.get(SenseiJobConfig.INDEX_PATH);
			shards = createShards(indexPath, numShards, conf);
			
			
		    // set the starting generation for each shard
		    // when a reduce task fails, a new reduce task
		    // has to know where to re-start
		    setShardGeneration(conf, shards);

		    Shard.setIndexShards(conf, shards);

		    // MapTask.MapOutputBuffer uses JobContext.IO_SORT_MB to decide its max buffer size
		    // (max buffer size = 1/2 * JobContext.IO_SORT_MB).
		    // Here we half-en JobContext.IO_SORT_MB because we use the other half memory to
		    // build an intermediate form/index in Combiner.
		    conf.setInt(MRJobConfig.IO_SORT_MB,  conf.getInt(MRJobConfig.IO_SORT_MB, 100) / 2);
		    
		    //always using compound file format to speed up;
		    conf.setBoolean(SenseiJobConfig.USE_COMPOUND_FILE, true);

		    // create the job configuration
		    JobConf jobConf = new JobConf(conf, MRClass);
		    jobConf.setJobName(MRClass.getName() + "_"+ System.currentTimeMillis());

		    // provided by application<commit_msg>fix the remote schema not found problem.<commit_after>package com.sensei.indexing.hadoop.job;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.NumberFormat;
import java.util.Arrays;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.util.StringUtils;
import org.apache.log4j.Logger;

import com.sensei.indexing.hadoop.keyvalueformat.IntermediateForm;
import com.sensei.indexing.hadoop.keyvalueformat.Shard;
import com.sensei.indexing.hadoop.map.SenseiMapper;
import com.sensei.indexing.hadoop.reduce.FileSystemDirectory;
import com.sensei.indexing.hadoop.reduce.IndexUpdateOutputFormat;
import com.sensei.indexing.hadoop.reduce.SenseiCombiner;
import com.sensei.indexing.hadoop.reduce.SenseiReducer;
import com.sensei.indexing.hadoop.util.LuceneUtil;
import com.sensei.indexing.hadoop.util.MRJobConfig;
import com.sensei.indexing.hadoop.util.SenseiJobConfig;

public class MapReduceJob extends Configured {

	private static final NumberFormat NUMBER_FORMAT = NumberFormat.getInstance();
	private static final Logger logger = Logger.getLogger(MapReduceJob.class);
	
	  public JobConf createJob(Class MRClass) throws IOException, URISyntaxException {
		    
		    Configuration conf = getConf();
		    Path[] inputPaths;
		    Path outputPath;
		    Shard[] shards = null;
			int numMapTasks = conf.getInt("mapreduce.job.maps", 2);
			int numShards = conf.getInt(SenseiJobConfig.NUM_SHARDS, 2);
//			inputPaths = FileInputFormat.getInputPaths(jobConf);
			
		    String dirs = conf.get("mapreduce.input.fileinputformat.inputdir", null);
		    logger.info("dirs:"+ dirs);
		    String [] list = StringUtils.split(dirs);
		    logger.info("length after split:"+ list.length);
		    inputPaths = new Path[list.length];
		    for (int i = 0; i < list.length; i++) {
		    	inputPaths[i] = new Path(StringUtils.unEscapeString(list[i]));
		    }
		    logger.info("path[0] is:" + inputPaths[0]);
		    	    
			outputPath = new Path(conf.get("mapreduce.output.fileoutputformat.outputdir"));
			String indexPath = conf.get(SenseiJobConfig.INDEX_PATH);
			shards = createShards(indexPath, numShards, conf);
			
			
		    // set the starting generation for each shard
		    // when a reduce task fails, a new reduce task
		    // has to know where to re-start
		    setShardGeneration(conf, shards);

		    Shard.setIndexShards(conf, shards);

		    // MapTask.MapOutputBuffer uses JobContext.IO_SORT_MB to decide its max buffer size
		    // (max buffer size = 1/2 * JobContext.IO_SORT_MB).
		    // Here we half-en JobContext.IO_SORT_MB because we use the other half memory to
		    // build an intermediate form/index in Combiner.
		    conf.setInt(MRJobConfig.IO_SORT_MB,  conf.getInt(MRJobConfig.IO_SORT_MB, 100) / 2);
		    
		    //always using compound file format to speed up;
		    conf.setBoolean(SenseiJobConfig.USE_COMPOUND_FILE, true);
		    
		    String schemaFile = conf.get("schema.file.url");
		    if(schemaFile == null)
		    	throw new IOException("no schema file is found");
		    else{
		    	logger.info("Adding schema file: " + conf.get("schema.file.url"));	      
				DistributedCache.addCacheFile(new URI(schemaFile), conf);
		    }

		    // create the job configuration
		    JobConf jobConf = new JobConf(conf, MRClass);
		    jobConf.setJobName(MRClass.getName() + "_"+ System.currentTimeMillis());

		    // provided by application
{'fraction_non_alphanumeric': 0.10885558583106267, 'fraction_numerical': 0.0032697547683923707, 'mean_word_length': 3.940107671601615, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7664312', 'n_tokens_mistral': 2499, 'n_tokens_neox': 2267, 'n_words': 476}