参考链接https://blog.csdn.net/zhou4411781/article/details/119133431配置src/main/java/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.javapublicstaticfinalString SPLIT_MAXSIZEmapreduce.input.fileinputformat.split.maxsize;publicstaticfinalString SPLIT_MINSIZEmapreduce.input.fileinputformat.split.minsize;propertynamemapreduce.input.fileinputformat.split.minsize/namevalue0/valuedescriptionThe minimum size chunk that map input should be split into.Note that some file formats may have minimum split sizes that take priority overthissetting./description/propertymapreduce.input.fileinputformat.split.maxsizesrc/main/java/org/apache/hadoop/mapreduce/lib/input/FileInputFormat.java/** * Get the maximum split size. * param context the job to look at. * return the maximum number of bytes a split can include */publicstaticlonggetMaxSplitSize(JobContext context){returncontext.getConfiguration().getLong(SPLIT_MAXSIZE,Long.MAX_VALUE);}/** * Generate the list of files and make them into FileSplits. * param job the job context * throws IOException */publicListInputSplitgetSplits(JobContext job)throws IOException{StopWatch swnewStopWatch().start();longminSizeMath.max(getFormatMinSplitSize(),getMinSplitSize(job));longmaxSizegetMaxSplitSize(job);src/main/java/org/apache/hadoop/mapred/FileInputFormat.javaif(length!0){FileSystem fspath.getFileSystem(job);BlockLocation[]blkLocations;if(file instanceof LocatedFileStatus){blkLocations((LocatedFileStatus)file).getBlockLocations();}else{blkLocationsfs.getFileBlockLocations(file,0,length);}if(isSplitable(fs,path)){longblockSizefile.getBlockSize();longsplitSizecomputeSplitSize(goalSize,minSize,blockSize);longbytesRemaininglength;while(((double)bytesRemaining)/splitSizeSPLIT_SLOP){String[][]splitHostsgetSplitHostsAndCachedHosts(blkLocations,length-bytesRemaining,splitSize,clusterMap);splits.add(makeSplit(path,length-bytesRemaining,splitSize,splitHosts[0],splitHosts[1]));bytesRemaining-splitSize;}示例2026-01-0815:26:46,125INFO mapred.FileInputFormat:Total input files to process:2562026-01-0815:26:46,157INFO mapreduce.JobSubmitter:number of splits:256Debugsrc/main/java/org/apache/hadoop/mapreduce/JobSubmitter.java//method to write splits for old api mapper.privateintwriteOldSplits(JobConf job,Path jobSubmitDir)throws IOException{org.apache.hadoop.mapred.InputSplit[]splitsjob.getInputFormat().getSplits(job,job.getNumMapTasks());// sort the splits into order based on size, so that the biggest// go firstArrays.sort(splits,newComparatororg.apache.hadoop.mapred.InputSplit(){publicintcompare(org.apache.hadoop.mapred.InputSplit a,org.apache.hadoop.mapred.InputSplit b){//method to write splits for old api mapper.privateintwriteOldSplits(JobConf job,Path jobSubmitDir)throws IOException{org.apache.hadoop.mapred.InputSplit[]splitsjob.getInputFormat().getSplits(job,job.getNumMapTasks());// sort the splits into order based on size, so that the biggest// go firstArrays.sort(splits,newComparatororg.apache.hadoop.mapred.InputSplit(){publicintcompare(org.apache.hadoop.mapred.InputSplit a,org.apache.hadoop.mapred.InputSplit b){try{wordcount debugsrc/main/java/org/apache/hadoop/examples/WordCount.javajob.setJarByClass(WordCount.class);job.setMapperClass(TokenizerMapper.class);job.setCombinerClass(IntSumReducer.class);job.setReducerClass(IntSumReducer.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);for(inti0;iotherArgs.length-1;i){FileInputFormat.addInputPath(job,newPath(otherArgs[i]));}FileOutputFormat.setOutputPath(job,newPath(otherArgs[otherArgs.length-1]));System.exit(job.waitForCompletion(true)?0:1);importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;这些属于 new apiold api 如下importorg.apache.hadoop.mapred.FileInputFormat;importorg.apache.hadoop.mapred.FileOutputFormat;importorg.apache.hadoop.mapred.JobClient;importorg.apache.hadoop.mapred.JobConf;importorg.apache.hadoop.mapred.Mapper;importorg.apache.hadoop.mapred.OutputCollector;importorg.apache.hadoop.mapred.Reporter;importorg.apache.hadoop.mapred.SequenceFileInputFormat;waitForCompletionsrc/main/java/org/apache/hadoop/mapreduce/Job.java/** * Submit the job to the cluster and wait for it to finish. * param verbose print the progress to the user * return true if the job succeeded * throws IOException thrown if the communication with the * codeJobTracker/code is lost */publicbooleanwaitForCompletion(boolean verbose)throws IOException,InterruptedException,ClassNotFoundException{if(stateJobState.DEFINE){submit();}/** * Submit the job to the cluster and return immediately. * throws IOException */publicvoidsubmit()throws IOException,InterruptedException,ClassNotFoundException{ensureState(JobState.DEFINE);setUseNewAPI();connect();/** * Default to the new APIs unless they are explicitly set or the old mapper or * reduce attributes are used. * throws IOException if the configuration is inconsistent */privatevoidsetUseNewAPI()throws IOException{intnumReducesconf.getNumReduceTasks();String oldMapperClassmapred.mapper.class;String oldReduceClassmapred.reducer.class;conf.setBooleanIfUnset(mapred.mapper.new-api,conf.get(oldMapperClass)null);if(conf.getUseNewMapper()){String modenew map API;src/main/java/org/apache/hadoop/mapreduce/JobSubmitter.javaprivateT extends InputSplitintwriteNewSplits(JobContext job,Path jobSubmitDir)throws IOException,InterruptedException,ClassNotFoundException{Configuration confjob.getConfiguration();InputFormat?,?inputReflectionUtils.newInstance(job.getInputFormatClass(),conf);ListInputSplitsplitsinput.getSplits(job);publicListInputSplitgetSplits(JobContext job)throws IOException{longminSizeNode0;longminSizeRack0;longmaxSize0;Configuration confjob.getConfiguration();// the values specified by setxxxSplitSize() takes precedence over the// values that might have been specified in the configif(minSplitSizeNode!0){minSizeNodeminSplitSizeNode;}else{minSizeNodeconf.getLong(SPLIT_MINSIZE_PERNODE,0);}src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.javapublicListInputSplitgetSplits(JobContext job)throws IOException{longminSizeNode0;longminSizeRack0;longmaxSize0;Configuration confjob.getConfiguration();// the values specified by setxxxSplitSize() takes precedence over the// values that might have been specified in the configif(minSplitSizeNode!0){minSizeNodeminSplitSizeNode;}else{minSizeNodeconf.getLong(SPLIT_MINSIZE_PERNODE,0);}if(minSplitSizeRack!0){minSizeRackminSplitSizeRack;}else{minSizeRackconf.getLong(SPLIT_MINSIZE_PERRACK,0);}if(maxSplitSize!0){maxSizemaxSplitSize;}else{maxSizeconf.getLong(mapreduce.input.fileinputformat.split.maxsize,0);// If maxSize is not configured, a single split will be generated per// node.}// create splits for all files that are not in any pool.getMoreSplits(job,stats,maxSize,minSizeNode,minSizeRack,splits);// free up rackToNodes maprackToNodes.clear();returnsplits;}/** * Return all the splits in the specified set of paths */privatevoidgetMoreSplits(JobContext job,ListFileStatusstats,longmaxSize,longminSizeNode,longminSizeRack,ListInputSplitsplits)throws IOException{Configuration confjob.getConfiguration();// all blocks for all the files in input setOneFileInfo[]files;// mapping from a rack name to the list of blocks it hasHashMapString,ListOneBlockInforackToBlocksnewHashMapString,ListOneBlockInfo();// mapping from a block to the nodes on which it has replicasHashMapOneBlockInfo,String[]blockToNodesnewHashMapOneBlockInfo,String[]();// mapping from a node to the list of blocks that it containsHashMapString,SetOneBlockInfonodeToBlocksnewHashMapString,SetOneBlockInfo();filesnewOneFileInfo[stats.size()];if(stats.size()0){return;}// populate all the blocks for all fileslongtotLength0;inti0;for(FileStatus stat:stats){files[i]newOneFileInfo(stat,conf,isSplitable(job,stat.getPath()),rackToBlocks,blockToNodes,nodeToBlocks,rackToNodes,maxSize);totLengthfiles[i].getLength();}createSplits(nodeToBlocks,blockToNodes,rackToBlocks,totLength,maxSize,minSizeNode,minSizeRack,splits);}split 相关截图两种合并策略echo 第 $i 次运行 $(date %Y-%m-%d %H:%M:%S)result-1536f-128m-am200-180m-128r.txt/usr/bin/time-p-a-o result-teragen.txt hadoop jar/usr/local/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.3.jar teragen-Dmapreduce.job.maps1536-Ddfs.blocksize128M16106127360/terasort/terasort-input sleep100#sshhadoop01echo 3 /proc/sys/vm/drop_caches#sshhadoop02echo 3 /proc/sys/vm/drop_caches#sshhadoop03echo 3 /proc/sys/vm/drop_caches#sleep50# 排序/usr/bin/time-p-a-o result-1536f-128m-am200-180m-128r.txt hadoop jar/usr/local/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.3.jar terasort \-Dmapreduce.task.io.sort.mb1280\-Dmapreduce.map.memory.mb3100\-Dmapreduce.reduce.memory.mb4375\-Dmapreduce.input.fileinputformat.split.minsize1073741824\-Dyarn.app.mapreduce.am.scheduler.heartbeat.interval-ms200\-Dmapreduce.job.reduces1170\-Dmapreduce.map.java.opts-noverify -XX:UseSerialGC -XX:MetaspaceSize64m -XX:CompileThreshold500 -XX:CICompilerCount2 -Djava.net.preferIPv4Stacktrue -Xmn1024m -Xms1536m -Xmx1536m -XX:UseTransparentHugePages\-Dmapreduce.reduce.java.opts-noverify -XX:UseSerialGC -XX:MetaspaceSize64m -XX:CompileThreshold500 -XX:CICompilerCount2 -Djava.net.preferIPv4Stacktrue -Xmn1024m -Xms1536m -Xmx1536m -XX:UseTransparentHugePages\/terasort/terasort-input/terasort/terasort-outputnohup-sort-1.5t-$i.txt21#TestDFSIO-D mapred.input.format.classorg.apache.hadoop.mapred.lib.CombineSequenceFileInputFormat \-D mapred.max.split.size1120#Terasort-Dmapreduce.input.fileinputformat.split.minsize1073741824\Terasort 合并的是同一个文件里的多个 blockTestDFSIO 合并的是多个独立小文件——前者 FileInputFormat (mapreduce.input.fileinputformat.split.minsize1073741824 )支持后者它根本做不到。
