public Job createJob ( FileSystemDataset dataset ) throws IOException { Configuration conf = HadoopUtils . getConfFromState ( state ) ; // Turn on mapreduce output compression by default if ( conf . get ( "mapreduce.output.fileoutputformat.compress" ) == null && conf . get ( "mapred.output.compress" ) == null ) { conf . setBoolean ( "mapreduce.output.fileoutputformat.compress" , true ) ; } // Disable delegation token cancellation by default if ( conf . get ( "mapreduce.job.complete.cancel.delegation.tokens" ) == null ) { conf . setBoolean ( "mapreduce.job.complete.cancel.delegation.tokens" , false ) ; } addJars ( conf , this . state , fs ) ; Job job = Job . getInstance ( conf ) ; job . setJobName ( MRCompactorJobRunner . HADOOP_JOB_NAME ) ; boolean emptyDirectoryFlag = this . configureInputAndOutputPaths ( job , dataset ) ; if ( emptyDirectoryFlag ) { this . state . setProp ( HiveRegistrationPolicy . MAPREDUCE_JOB_INPUT_PATH_EMPTY_KEY , true ) ; } this . configureMapper ( job ) ; this . configureReducer ( job ) ; if ( emptyDirectoryFlag || ! this . shouldDeduplicate ) { job . setNumReduceTasks ( 0 ) ; } // Configure schema at the last step because FilesInputFormat will be used internally this . configureSchema ( job ) ; this . isJobCreated = true ; this . configuredJob = job ; return job ; }
Customized MR job creation for Avro .