MapReduce项目案例1

统计单词数 1.输入内容

hello world i am teacher
hello world i am teacher
hello world i am teacher
hello world i am teacher
hello world
hello world

2.输出内容

am	4
hello	6
i	4
teacher	4
world	6

3.代码

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;

import java.io.IOException;
import java.util.StringTokenizer;

public class WordCount {

    
    public static class TokenizerMapper
            extends Mapper {//<前两位表示输入类型<偏移量，一行文本>，后两位表示输出类型<单词，数字>>

        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

        
        @Override
        public void map(Object key, Text value, Context context
        ) throws IOException, InterruptedException {
            System.out.println("切分split后数据--偏移量:" + key + "t值:" + value);
            StringTokenizer itr = new StringTokenizer(value.toString());//根据自然分隔符分割
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());//写入文本对象
                context.write(word, one);//保存出去(单词，数字)
            }
        }
    }

    
    public static class IntSumReducer
            extends Reducer {
        private IntWritable result = new IntWritable();

        
        @Override
        public void reduce(Text key, Iterable 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();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
        Job job = Job.getInstance(conf, "word count");//名字
        job.setJarByClass(WordCount.class);//上线的jar
        job.setMapperClass(TokenizerMapper.class);//mapper
        job.setCombinerClass(IntSumReducer.class);//combine:合并一个节点
        job.setReducerClass(IntSumReducer.class);//reduce:合并不同节点
        job.setOutputKeyClass(Text.class);//输出键的类型，与上面一致
        job.setOutputValueClass(IntWritable.class);//输出值的类型，与上面一致
        FileInputFormat.addInputPath(job, new Path("E:\HadoopMRData\input"));//输入目录
        FileOutputFormat.setOutputPath(job, new Path("E:\HadoopMRData\output"));//输出目录，这个目录要不存在运行时创建，特别注意要设置到空目录，应为执行之前会删一次****
       
        System.exit(job.waitForCompletion(true) ? 0 : 1);//启动，0表示正常退出
    }
}

去除重复项 1.输入内容

2.输出内容

11	
12	
13

3.代码

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class RemoveRepeat {

    
    public static class MyMapper extends Mapper {

        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            System.out.println("切分split后数据--偏移量:" + key + "t值:" + value);
            context.write(new Text(value), new Text(""));
        }

    }

    

    public static class MyReducer extends Reducer {
        @Override
        public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            context.write(key, new Text(""));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
        Job job = Job.getInstance(conf, "remove repeat");//名字
        job.setJarByClass(RemoveRepeat.class);//上线的jar
        job.setMapperClass(MyMapper.class);//mapper
        job.setCombinerClass(MyReducer.class);//combine:合并一个节点
        job.setReducerClass(MyReducer.class);//reduce:合并不同节点
        job.setOutputKeyClass(Text.class);//输出键的类型，与上面一致
        job.setOutputValueClass(Text.class);//输出值的类型，与上面一致
        FileInputFormat.addInputPath(job, new Path("E:\HadoopMRData\input"));//输入目录
        FileOutputFormat.setOutputPath(job, new Path("E:\HadoopMRData\output"));//输出目录，这个目录要不存在运行时创建，特别注意要设置到空目录，应为执行之前会删一次****
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);//启动，0表示正常退出
    }
}

去重并两种方式输出序号 1.输入内容

2.输出内容

3.代码

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;

import java.io.IOException;

public class Order {

    public static class MyMapper extends Mapper {
        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            System.out.println("切分split后数据--偏移量:" + key + "t值:" + value);
            int va = Integer.parseInt(value.toString());
            context.write(new IntWritable(va), new Text("1"));
        }
    }

    

    public static class MyReducer extends Reducer {
        private int num1 = 1;//负责连续序号
        private int num2 = 1;//负责跳跃序号

        @Override
        protected void reduce(IntWritable key, Iterable values, Context context) throws IOException, InterruptedException {
            context.write(new IntWritable(num1), new Text(num2 + "t" + key.get()));
            for (Text val : values) {
                num2++;
            }
            num1++;
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
        Job job = Job.getInstance(conf, "order");//名字
        job.setJarByClass(Order.class);//上线的jar
        job.setMapperClass(MyMapper.class);//mapper


        
        //job.setCombinerClass(MyReducer.class);//combine:合并一个节点


        job.setReducerClass(MyReducer.class);//reduce:合并不同节点
        job.setOutputKeyClass(IntWritable.class);//输出键的类型，与上面一致
        job.setOutputValueClass(Text.class);//输出值的类型，与上面一致
        FileInputFormat.addInputPath(job, new Path("E:\HadoopMRData\input"));//输入目录
        FileOutputFormat.setOutputPath(job, new Path("E:\HadoopMRData\output"));//输出目录，这个目录要不存在运行时创建，特别注意要设置到空目录，应为执行之前会删一次****

        
        System.exit(job.waitForCompletion(true) ? 0 : 1);//启动，0表示正常退出
    }
}

同一个学生的平均值 1.输入内容

张三	98
李四	94
王五	89
张三	86
李四	92
王五	86
张三	82
李四	90

2.输出内容

张三	88.66666666666667
李四	89.5
王五	89.54166666666667

3.代码

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;

public class Average {

    public static class MyMapper extends Mapper {
        @Override
        public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            System.out.println("切分split后数据--偏移量:" + key + "t值:" + value);
            StringTokenizer tokenizer = new StringTokenizer(value.toString());//根据自然分割符分割
            while (tokenizer.hasMoreTokens()) {
                context.write(new Text(tokenizer.nextToken()), new Text(tokenizer.nextToken()));
            }
        }
    }

    

    public static class MyReducer extends Reducer {
        private int count = 0;//记录该人有多少成绩
        private double sum = 0;//记录成绩总和

        @Override
        protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            for (Text val : values) {
                sum += Double.parseDouble((val.toString()));
                count++;
            }
            context.write(key, new Text(String.valueOf(sum / count)));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();//配置参数(map数,reduce数...对于配置配置文件中的API配置)
        Job job = Job.getInstance(conf, "average");//名字
        job.setJarByClass(Average.class);//上线的jar
        job.setMapperClass(MyMapper.class);//mapper
        job.setCombinerClass(MyReducer.class);//combine:合并一个节点
        job.setReducerClass(MyReducer.class);//reduce:合并不同节点
        job.setOutputKeyClass(Text.class);//输出键的类型，与上面一致
        job.setOutputValueClass(Text.class);//输出值的类型，与上面一致
        FileInputFormat.addInputPath(job, new Path("E:\HadoopMRData\input"));//输入目录
        FileOutputFormat.setOutputPath(job, new Path("E:\HadoopMRData\output"));//输出目录，这个目录要不存在运行时创建，特别注意要设置到空目录，应为执行之前会删一次****

        
        System.exit(job.waitForCompletion(true) ? 0 : 1);//启动，0表示正常退出
    }
}

MapReduce项目案例1

Java相关栏目本月热门文章