public static void writePartitionFile ( JobConf conf , Path partFile ) throws IOException { TeraInputFormat inFormat = new TeraInputFormat ( ) ; TextSampler sampler = new TextSampler ( ) ; Text key = new Text ( ) ; Text value = new Text ( ) ; int partitions = conf . getNumReduceTasks ( ) ; long sampleSize = conf . getLong ( SAMPLE_SIZE , 100000 ) ; InputSplit [ ] splits = inFormat . getSplits ( conf , conf . getNumMapTasks ( ) ) ; int samples = Math . min ( 10 , splits . length ) ; long recordsPerSample = sampleSize / samples ; int sampleStep = splits . length / samples ; long records = 0 ; // take N samples from different parts of the input for ( int i = 0 ; i < samples ; ++ i ) { RecordReader < Text , Text > reader = inFormat . getRecordReader ( splits [ sampleStep * i ] , conf , null ) ; while ( reader . next ( key , value ) ) { sampler . addKey ( key ) ; records += 1 ; if ( ( i + 1 ) * recordsPerSample <= records ) { break ; } } } FileSystem outFs = partFile . getFileSystem ( conf ) ; if ( outFs . exists ( partFile ) ) { outFs . delete ( partFile , false ) ; } SequenceFile . Writer writer = SequenceFile . createWriter ( outFs , conf , partFile , Text . class , NullWritable . class ) ; NullWritable nullValue = NullWritable . get ( ) ; for ( Text split : sampler . createPartitions ( partitions ) ) { writer . append ( split , nullValue ) ; } writer . close ( ) ; }
Use the input splits to take samples of the input and generate sample keys . By default reads 100 000 keys from 10 locations in the input sorts them and picks N - 1 keys to generate N equally sized partitions .