<commit_before>package com.ibm.bi.dml.runtime.controlprogram.parfor;

import java.io.IOException;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Counters;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.lib.NLineInputFormat;

import com.ibm.bi.dml.lops.runtime.RunMRJobs.ExecMode;
import com.ibm.bi.dml.runtime.controlprogram.LocalVariableMap;
import com.ibm.bi.dml.runtime.controlprogram.ParForProgramBlock;
import com.ibm.bi.dml.runtime.controlprogram.caching.MatrixObject;
import com.ibm.bi.dml.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
import com.ibm.bi.dml.runtime.controlprogram.parfor.stat.Stat;
import com.ibm.bi.dml.runtime.instructions.CPInstructions.Data;
import com.ibm.bi.dml.runtime.matrix.MatrixCharacteristics;
import com.ibm.bi.dml.runtime.matrix.MatrixFormatMetaData;
import com.ibm.bi.dml.runtime.matrix.mapred.MRJobConfiguration;
import com.ibm.bi.dml.runtime.util.MapReduceTool;
import com.ibm.bi.dml.utils.DMLRuntimeException;
import com.ibm.bi.dml.utils.Statistics;

/**
 * MR job class for submitting parfor remote MR jobs, controlling its execution and obtaining results.
 * 
 *
 */
public class RemoteParForMR
{
	/**
	 * 
	 * @param pfid
	 * @param program
	 * @param taskFile
	 * @param resultFile
	 * @param mode
	 * @param numMappers
	 * @param replication
	 * @return
	 * @throws DMLRuntimeException
	 */
	public static RemoteParForJobReturn runJob(long pfid, String program, String taskFile, String resultFile, MatrixObject colocatedDPMatrixObj, //inputs
			                       ExecMode mode, int numMappers, int replication, int max_retry, long minMem)  //opt params
		throws DMLRuntimeException
	{
		RemoteParForJobReturn ret = null;
		
		JobConf job;
		job = new JobConf( RemoteParForMR.class );
		job.setJobName("ParFor_Execute-MR"+pfid);
		
		//maintain dml script counters
		Statistics.incrementNoOfCompiledMRJobs();
	
		try
		{
			/////
			//configure the MR job
		
			//set arbitrary CP program blocks that will perform in the mapper
			MRJobConfiguration.setProgramBlocksInMapper(job, program); 
			
			//set mappers, reducers, combiners
			job.setMapperClass(RemoteParWorkerMapper.class); //map-only

			//set input format (one split per row, NLineInputFormat default N=1)
			if( ParForProgramBlock.ALLOW_DATA_COLOCATION && colocatedDPMatrixObj != null )
			{
				job.setInputFormat(RemoteParForColocatedNLineInputFormat.class);
				MRJobConfiguration.setPartitioningFormat(job, colocatedDPMatrixObj.getPartitionFormat());
				MatrixCharacteristics mc = ((MatrixFormatMetaData)colocatedDPMatrixObj.getMetaData()).getMatrixCharacteristics();
				MRJobConfiguration.setPartitioningBlockNumRows(job, mc.get_rows_per_block());
				MRJobConfiguration.setPartitioningBlockNumCols(job, mc.get_cols_per_block());
				MRJobConfiguration.setPartitioningFilename(job, colocatedDPMatrixObj.getFileName());
			}
			else //default case 
			{
				job.setInputFormat(NLineInputFormat.class);
			}
			
			//set the input path and output path 
		    FileInputFormat.setInputPaths(job, new Path(taskFile));
			
		    //set output format
		    job.setOutputFormat(SequenceFileOutputFormat.class);
		    
		    //set output path
		    MapReduceTool.deleteFileIfExistOnHDFS(resultFile);
		    FileOutputFormat.setOutputPath(job, new Path(resultFile));
		    
			//set the output key, value schema
			job.setMapOutputKeyClass(LongWritable.class);
			job.setMapOutputValueClass(Text.class);
			job.setOutputKeyClass(LongWritable.class);
			job.setOutputValueClass(Text.class);
			
			//////
			//set optimization parameters

			//set the number of mappers and reducers 
			job.setNumMapTasks(numMappers); //numMappers
			job.setNumReduceTasks( 0 );			
			//job.setInt("mapred.map.tasks.maximum", 1); //system property
			//job.setInt("mapred.tasktracker.tasks.maximum",1); //system property
			//job.setInt("mapred.jobtracker.maxtasks.per.job",1); //system property

			//use FLEX scheduler configuration properties
			//System.out.println("numMappers="+numMappers);
			if( ParForProgramBlock.USE_FLEX_SCHEDULER_CONF )
			{
				job.setInt("flex.map.min", 0);
				job.setInt("flex.map.max", numMappers);
				job.setInt("flex.reduce.min", 0);
				job.setInt("flex.reduce.max", numMappers);
			}
			
			//set jvm memory size (if require)
			String memKey = "mapred.child.java.opts";
			if( minMem > 0 && minMem > InfrastructureAnalyzer.extractMaxMemoryOpt(job.get(memKey)) )
			{
				InfrastructureAnalyzer.setMaxMemoryOpt(job, memKey, minMem);
				System.out.println("Warning: Forcing '"+memKey+"' to -Xmx"+minMem/(1024*1024)+"M." );
			}
			
			//disable automatic tasks timeouts and speculative task exec
			job.setInt("mapred.task.timeout", 0);			
			job.setMapSpeculativeExecution(false);
			
			//enables the reuse of JVMs (multiple tasks per MR task)
			if( ParForProgramBlock.ALLOW_REUSE_MR_JVMS )
				job.setNumTasksToExecutePerJvm(-1); //unlimited
			
			//set the replication factor for the results
			job.setInt("dfs.replication", replication);
			
			//set the max number of retries per map task
			job.setInt("mapreduce.map.maxattempts", max_retry);
			
			// By default, the job executes in "cluster" mode.
			// Determine if we can optimize and run it in "local" mode.
			if ( mode == ExecMode.LOCAL ) {
				job.set("mapred.job.tracker", "local");	
				MRJobConfiguration.setStagingDir( job );
			}

			
			//set unique working dir
			MRJobConfiguration.setUniqueWorkingDir(job, mode);
			
			
			/////
			// execute the MR job			
			RunningJob runjob = JobClient.runJob(job);
			
			// Process different counters 
			Statistics.incrementNoOfExecutedMRJobs();
			Counters group = runjob.getCounters();
			int numTasks = (int)group.getCounter( Stat.PARFOR_NUMTASKS );
			int numIters = (int)group.getCounter( Stat.PARFOR_NUMITERS );
				
			// read all files of result variables and prepare for return
			LocalVariableMap[] results = readResultFile(resultFile); 

			ret = new RemoteParForJobReturn(runjob.isSuccessful(), 
					                        numTasks, numIters, 
					                        results);  	
		}
		catch(Exception ex)
		{
			throw new DMLRuntimeException(ex);
		}
		finally
		{
			// remove created files 
			try
			{
				MapReduceTool.deleteFileIfExistOnHDFS(new Path(taskFile), job);
				MapReduceTool.deleteFileIfExistOnHDFS(new Path(resultFile), job);
			}
			catch(IOException ex)
			{
				throw new DMLRuntimeException(ex);
			}
		}
		
		return ret;
	}
	

	/**
	 * 
	 * @param fname
	 * @return
	 * @throws DMLRuntimeException
	 */
	public static LocalVariableMap [] readResultFile(String fname)
		throws DMLRuntimeException, IOException
	{
		LocalVariableMap [] ret = null;
		HashMap<Long,LocalVariableMap> tmp = new HashMap<Long,LocalVariableMap>();

		SequenceFile.Reader reader = null;
		
		Path path = new Path(fname);
		Configuration conf = new Configuration();
		FileSystem fs = FileSystem.get(conf);
		
		//foreach file in path
		FileStatus[] status = fs.listStatus(path);
		for( FileStatus f : status )
		{
			//System.out.println("Result filepath: "+f.getPath().getName());
			if( f.getPath().getName().startsWith("_") ) //reject system-internal filenames
				continue;
				
			try
			{
				reader = new SequenceFile.Reader(fs, f.getPath(), conf);
				
				LongWritable key = new LongWritable();
		        Text value = new Text();
		
		        while( reader.next(key, value) ) 
		        {
		        	if( !tmp.containsKey( key.get() ) )
		        		tmp.put(key.get(), new LocalVariableMap ());	        	
		        	Object[] dat = ProgramConverter.parseDataObject( value.toString() );
		        	tmp.get( key.get() ).put((String)dat[0], (Data)dat[1]);
		        }
			}
			catch(Exception ex)
			{
				throw new DMLRuntimeException("Error reading results from resultfile "+fname, ex);
			}
			finally
			{
		        if(reader!=null)
		        	reader.close();				
			}
		}

		
		//create return
		ret = tmp.values().toArray(new LocalVariableMap[0]);	
			
		return ret;
	}
}<commit_msg>20974: ParFor construct - Cleanup: read result files from remote workers<commit_after>package com.ibm.bi.dml.runtime.controlprogram.parfor;

import java.io.IOException;
import java.util.HashMap;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.Counters;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.InputSplit;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.lib.NLineInputFormat;

import com.ibm.bi.dml.lops.runtime.RunMRJobs.ExecMode;
import com.ibm.bi.dml.runtime.controlprogram.LocalVariableMap;
import com.ibm.bi.dml.runtime.controlprogram.ParForProgramBlock;
import com.ibm.bi.dml.runtime.controlprogram.caching.MatrixObject;
import com.ibm.bi.dml.runtime.controlprogram.parfor.stat.InfrastructureAnalyzer;
import com.ibm.bi.dml.runtime.controlprogram.parfor.stat.Stat;
import com.ibm.bi.dml.runtime.instructions.CPInstructions.Data;
import com.ibm.bi.dml.runtime.matrix.MatrixCharacteristics;
import com.ibm.bi.dml.runtime.matrix.MatrixFormatMetaData;
import com.ibm.bi.dml.runtime.matrix.mapred.MRJobConfiguration;
import com.ibm.bi.dml.runtime.util.MapReduceTool;
import com.ibm.bi.dml.utils.DMLRuntimeException;
import com.ibm.bi.dml.utils.Statistics;

/**
 * MR job class for submitting parfor remote MR jobs, controlling its execution and obtaining results.
 * 
 *
 */
public class RemoteParForMR
{
	/**
	 * 
	 * @param pfid
	 * @param program
	 * @param taskFile
	 * @param resultFile
	 * @param mode
	 * @param numMappers
	 * @param replication
	 * @return
	 * @throws DMLRuntimeException
	 */
	public static RemoteParForJobReturn runJob(long pfid, String program, String taskFile, String resultFile, MatrixObject colocatedDPMatrixObj, //inputs
			                       ExecMode mode, int numMappers, int replication, int max_retry, long minMem)  //opt params
		throws DMLRuntimeException
	{
		RemoteParForJobReturn ret = null;
		
		JobConf job;
		job = new JobConf( RemoteParForMR.class );
		job.setJobName("ParFor_Execute-MR"+pfid);
		
		//maintain dml script counters
		Statistics.incrementNoOfCompiledMRJobs();
	
		try
		{
			/////
			//configure the MR job
		
			//set arbitrary CP program blocks that will perform in the mapper
			MRJobConfiguration.setProgramBlocksInMapper(job, program); 
			
			//set mappers, reducers, combiners
			job.setMapperClass(RemoteParWorkerMapper.class); //map-only

			//set input format (one split per row, NLineInputFormat default N=1)
			if( ParForProgramBlock.ALLOW_DATA_COLOCATION && colocatedDPMatrixObj != null )
			{
				job.setInputFormat(RemoteParForColocatedNLineInputFormat.class);
				MRJobConfiguration.setPartitioningFormat(job, colocatedDPMatrixObj.getPartitionFormat());
				MatrixCharacteristics mc = ((MatrixFormatMetaData)colocatedDPMatrixObj.getMetaData()).getMatrixCharacteristics();
				MRJobConfiguration.setPartitioningBlockNumRows(job, mc.get_rows_per_block());
				MRJobConfiguration.setPartitioningBlockNumCols(job, mc.get_cols_per_block());
				MRJobConfiguration.setPartitioningFilename(job, colocatedDPMatrixObj.getFileName());
			}
			else //default case 
			{
				job.setInputFormat(NLineInputFormat.class);
			}
			
			//set the input path and output path 
		    FileInputFormat.setInputPaths(job, new Path(taskFile));
			
		    //set output format
		    job.setOutputFormat(SequenceFileOutputFormat.class);
		    
		    //set output path
		    MapReduceTool.deleteFileIfExistOnHDFS(resultFile);
		    FileOutputFormat.setOutputPath(job, new Path(resultFile));
		    
			//set the output key, value schema
			job.setMapOutputKeyClass(LongWritable.class);
			job.setMapOutputValueClass(Text.class);			
			job.setOutputKeyClass(LongWritable.class);
			job.setOutputValueClass(Text.class);
			
			//////
			//set optimization parameters

			//set the number of mappers and reducers 
			job.setNumMapTasks(numMappers); //numMappers
			job.setNumReduceTasks( 0 );			
			//job.setInt("mapred.map.tasks.maximum", 1); //system property
			//job.setInt("mapred.tasktracker.tasks.maximum",1); //system property
			//job.setInt("mapred.jobtracker.maxtasks.per.job",1); //system property

			//use FLEX scheduler configuration properties
			//System.out.println("numMappers="+numMappers);
			if( ParForProgramBlock.USE_FLEX_SCHEDULER_CONF )
			{
				job.setInt("flex.map.min", 0);
				job.setInt("flex.map.max", numMappers);
				job.setInt("flex.reduce.min", 0);
				job.setInt("flex.reduce.max", numMappers);
			}
			
			//set jvm memory size (if require)
			String memKey = "mapred.child.java.opts";
			if( minMem > 0 && minMem > InfrastructureAnalyzer.extractMaxMemoryOpt(job.get(memKey)) )
			{
				InfrastructureAnalyzer.setMaxMemoryOpt(job, memKey, minMem);
				System.out.println("Warning: Forcing '"+memKey+"' to -Xmx"+minMem/(1024*1024)+"M." );
			}
			
			//disable automatic tasks timeouts and speculative task exec
			job.setInt("mapred.task.timeout", 0);			
			job.setMapSpeculativeExecution(false);
			
			//enables the reuse of JVMs (multiple tasks per MR task)
			if( ParForProgramBlock.ALLOW_REUSE_MR_JVMS )
				job.setNumTasksToExecutePerJvm(-1); //unlimited
			
			//set the replication factor for the results
			job.setInt("dfs.replication", replication);
			
			//set the max number of retries per map task
			job.setInt("mapreduce.map.maxattempts", max_retry);
			
			// By default, the job executes in "cluster" mode.
			// Determine if we can optimize and run it in "local" mode.
			if ( mode == ExecMode.LOCAL ) {
				job.set("mapred.job.tracker", "local");	
				MRJobConfiguration.setStagingDir( job );
			}
			
			//set unique working dir
			MRJobConfiguration.setUniqueWorkingDir(job, mode);
			
			
			/////
			// execute the MR job			
			RunningJob runjob = JobClient.runJob(job);
			
			// Process different counters 
			Statistics.incrementNoOfExecutedMRJobs();
			Counters group = runjob.getCounters();
			int numTasks = (int)group.getCounter( Stat.PARFOR_NUMTASKS );
			int numIters = (int)group.getCounter( Stat.PARFOR_NUMITERS );
				
			// read all files of result variables and prepare for return
			LocalVariableMap[] results = readResultFile(resultFile); 

			ret = new RemoteParForJobReturn(runjob.isSuccessful(), 
					                        numTasks, numIters, 
					                        results);  	
		}
		catch(Exception ex)
		{
			throw new DMLRuntimeException(ex);
		}
		finally
		{
			// remove created files 
			try
			{
				MapReduceTool.deleteFileIfExistOnHDFS(new Path(taskFile), job);
				MapReduceTool.deleteFileIfExistOnHDFS(new Path(resultFile), job);
			}
			catch(IOException ex)
			{
				throw new DMLRuntimeException(ex);
			}
		}
		
		return ret;
	}
	

	/**
	 * 
	 * @param fname
	 * @return
	 * @throws DMLRuntimeException
	 */
	public static LocalVariableMap [] readResultFile( String fname )
		throws DMLRuntimeException, IOException
	{
		HashMap<Long,LocalVariableMap> tmp = new HashMap<Long,LocalVariableMap>();

		JobConf job = new JobConf();
		Path path = new Path(fname);
		FileInputFormat.addInputPath(job, path); 
		
		SequenceFileInputFormat<LongWritable,Text> informat = new SequenceFileInputFormat<LongWritable,Text>();
		InputSplit[] splits = informat.getSplits(job, 1);
		LongWritable key = new LongWritable();
		Text value = new Text();
		
		for(InputSplit split: splits)
		{
			RecordReader<LongWritable,Text> reader = informat.getRecordReader(split, job, Reporter.NULL);
			try
			{
				while( reader.next(key, value) )
				{
					if( !tmp.containsKey( key.get() ) )
		        		tmp.put(key.get(), new LocalVariableMap ());	   
					Object[] dat = ProgramConverter.parseDataObject( value.toString() );
		        	tmp.get( key.get() ).put((String)dat[0], (Data)dat[1]);
				}
			}	
			finally
			{
				if( reader != null )
					reader.close();
			}
		}		

		//create return array
		return tmp.values().toArray(new LocalVariableMap[0]);	
	}
}
{'fraction_non_alphanumeric': 0.11384578800309138, 'fraction_numerical': 0.0028535758872837527, 'mean_word_length': 3.630333058078723, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 14, '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': '4558765', 'n_tokens_mistral': 6203, 'n_tokens_neox': 5309, 'n_words': 1052}