MapReduce练习

清疚 2023-10-08 15:23 25阅读 0赞

### MapAction ###

package com.item.test;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    public class MapAction extends Mapper<LongWritable, Text, Text, LongWritable> {
        
    
        // 定义输出outk
        private Text outk = new Text();
        // 定义输出outv
        private LongWritable outv = new LongWritable(1);
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        
            // 1.读取一行数据，将Text转为Char数据类型
            char[] line = value.toString().toCharArray();
    
            // 2.将line数据循环写出
            for (char c : line) {
        
                outk.set(c+"");
                context.write(outk, outv);
            }
        }
    }

### ReduceAction ###

package com.item.test;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class ReduceAction extends Reducer<Text, LongWritable, Text, LongWritable> {
        
    
        // 定义输出outv
        private LongWritable outv = new LongWritable();
    
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
        
            // 1.将values数据，循环迭代汇总value
            long count = 0;
            for (LongWritable value : values) {
        
                count += value.get();
            }
            // 2.封装outv
            outv.set(count);
            // 3.提交数据
            context.write(key, outv);
        }
    }

### Action ###

package com.item.test;
    
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    
    import java.io.IOException;
    
    public class Action {
        
    
        public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        
    
            // 1.获取配置信息、Job对象实例
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
    
            // 2.设置jar包路径
            job.setJarByClass(Action.class);
    
            // 3.关联mapper和reducer
            job.setMapperClass(MapAction.class);
            job.setReducerClass(ReduceAction.class);
    
            // 4.设置map输出的kv类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);
    
            // 5.设置最终输出的kv类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);
    
            // 6.设置输入、输出路径
            FileInputFormat.setInputPaths(job, new Path("/info/info.txt"));
            FileOutputFormat.setOutputPath(job, new Path("/infos"));
    
            // 7.提交job
            boolean result = job.waitForCompletion(true);
            System.exit(result ? 0 : 1);
        }
    }