更新時間:2024-02-26 來源:黑馬程序員 瀏覽量:
在Hadoop中,Combiner的作用是在Map階段輸出數(shù)據(jù)之后,但在數(shù)據(jù)傳輸?shù)絉educer之前,對Map輸出的數(shù)據(jù)進(jìn)行一次局部聚合操作。Combiner可以大大減少Map階段輸出的數(shù)據(jù)量,從而減輕Reducer的負(fù)擔(dān),提高作業(yè)的整體性能。
Combiner通常用于對具有可結(jié)合性和可交換性的操作進(jìn)行局部合并,比如求和、計數(shù)等。它們在Map任務(wù)的輸出上運(yùn)行,將相同鍵的值合并到一起,以減少數(shù)據(jù)傳輸。
下面是一個簡單的示例,演示如何在Hadoop MapReduce作業(yè)中使用Combiner:
import java.io.IOException; import java.util.StringTokenizer; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class WordCount { public static class TokenizerMapper extends Mapper<LongWritable, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(LongWritable key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "word count"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); // 設(shè)置Combiner job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }
假設(shè)我們有一個文本文件,其中包含一系列單詞,我們想要計算每個單詞出現(xiàn)的次數(shù)。
在上面的示例中,我們定義了一個簡單的WordCount作業(yè)。在main函數(shù)中,我們使用job.setCombinerClass(IntSumReducer.class)來指定使用IntSumReducer作為Combiner。
在IntSumReducer類中,reduce函數(shù)負(fù)責(zé)將相同鍵的值相加,這是一個可結(jié)合的操作。通過將IntSumReducer作為Combiner,可以在Map階段對輸出的鍵值對進(jìn)行局部合并,減少數(shù)據(jù)傳輸量。