public RunningJob submitJobInternal ( JobConf job ) throws FileNotFoundException , ClassNotFoundException , InterruptedException , IOException { /* * configure the command line options correctly on the submitting dfs */ boolean shared = job . getBoolean ( "mapred.cache.shared.enabled" , false ) ; JobID jobId = jobSubmitClient . getNewJobId ( ) ; Path submitJobDir = new Path ( getSystemDir ( ) , jobId . toString ( ) ) ; Path sharedFilesDir = new Path ( getSystemDir ( ) , jobSubmitClient . CAR ) ; Path submitSplitFile = new Path ( submitJobDir , "job.split" ) ; getFs ( ) ; if ( jobSubmitClient instanceof LocalJobRunner ) { symLinkAndConfigureFiles ( job ) ; } else { copyAndConfigureFiles ( job , ( shared ) ? sharedFilesDir : submitJobDir , shared ) ; } Path submitJobFile = new Path ( submitJobDir , "job.xml" ) ; int reduces = job . getNumReduceTasks ( ) ; JobContext context = new JobContext ( job , jobId ) ; // Check the output specification if ( reduces == 0 ? job . getUseNewMapper ( ) : job . getUseNewReducer ( ) ) { org . apache . hadoop . mapreduce . OutputFormat < ? , ? > output = ReflectionUtils . newInstance ( context . getOutputFormatClass ( ) , job ) ; output . checkOutputSpecs ( context ) ; } else { job . getOutputFormat ( ) . checkOutputSpecs ( fs , job ) ; } // Create the splits for the job LOG . debug ( "Creating splits at " + fs . makeQualified ( submitSplitFile ) ) ; List < RawSplit > maps ; if ( job . getUseNewMapper ( ) ) { maps = computeNewSplits ( context ) ; } else { maps = computeOldSplits ( job ) ; } job . setNumMapTasks ( maps . size ( ) ) ; if ( ! isJobTrackerInProc ) { JobConf conf = null ; if ( job . getUseNewMapper ( ) ) { conf = context . getJobConf ( ) ; } else { conf = job ; } writeComputedSplits ( conf , maps , submitSplitFile ) ; job . set ( "mapred.job.split.file" , submitSplitFile . toString ( ) ) ; } else { synchronized ( JobClient . jobSplitCache ) { if ( JobClient . jobSplitCache . containsKey ( jobId ) ) { throw new IOException ( "Job split already cached " + jobId ) ; } JobClient . jobSplitCache . put ( jobId , maps ) ; } synchronized ( JobClient . jobConfCache ) { if ( JobClient . jobConfCache . containsKey ( jobId ) ) { throw new IOException ( "Job conf already cached " + jobId ) ; } jobConfCache . put ( jobId , job ) ; } } // Write job file to JobTracker's fs FSDataOutputStream out = FileSystem . create ( fs , submitJobFile , new FsPermission ( JOB_FILE_PERMISSION ) ) ; try { job . writeXml ( out ) ; } finally { out . close ( ) ; } // // Now, actually submit the job (using the submit name) // JobStatus status = jobSubmitClient . submitJob ( jobId ) ; if ( status != null ) { return new NetworkedJob ( status ) ; } else { throw new IOException ( "Could not launch job" ) ; } }
Internal method for submitting jobs to the system .