Hadoop系列-MapReduce小例子

2022-06-09 本文已影响0人汤太咸啊

Spark卡壳了，有点费劲，先把MapReduce学回来才开始Spark，基本上Spark是替换MapReduce的东西，所以先学习了MapReduce的原理，下面咱们开始学习哈。

MapReduce有三部分，一部分是Map和Reduce，另一个就是主执行方法，也就是启动类。

一、Map

Map部分主要是将各种文件的信息统计映射到具体的key/value键值对，也就是对信息的整理部分，统一规则，以便为接下来Reduce工作提供统一的数据格式。

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
// Mapper后边分别按顺序范型指定了输入key，输入value，输出key，输出value类型
// 由于咱们是文件读取输入，只需要处理输入value，输入key类型无所谓，Long就可以了
public  class Map extends Mapper<LongWritable, Text, Text, IntWritable> {
    // 输入key输入value具体值，全局的一个Context（方便传递数据给Reduce）
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        //文件的数据通过空格截取成单独的文字
        String line = value.toString();
        String[] words = line.split(" ");
        // 将文件中截取出来的各种字符循环写入到context中，以便Reduce继续处理
        // 实际就是将字符输出成key/value格式
        // 例如文件有hello world，将会写成key:hello,value:1;key:world,value:1的格式传递给Reduce
        for (String word : words) {
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

二、Reduce

Reduce部分主要是将整理好的key/value键值对进行统计计算处理，最终输出数据。

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
// Reducer后边范型指定了输入key，value，输出key，value的具体类型
// 注意这个时候的输入key，value要与前面的Map的输出key，value的具体类型对应
public class Reduce extends Reducer<Text, IntWritable, Text, LongWritable> {
    // 输入的key，values，以及全局的Context（方便最终输出文件）
    // 这个时候注意values是集合，Reduce是将前面Map输出的所有相同的key整合到一起
    // values就是这个整合的集合，对这个集合处理
    // 同一个key只执行Reduce方法一次
    @Override
    public void reduce(Text key,
                       Iterable<IntWritable> values,
                       Context context)
            throws IOException, InterruptedException {
        long count = 0;
        // 同一个key的value集合循环处理增加技术
        for (IntWritable value : values) {
            count++;
        }
        // 输出最终key的总数
        context.write(key, new LongWritable(count));
    }
}

三、Main执行方法

这个是具体的MapReduce的启动方法，后期打包后，hadoop命令执行该包之后，会从该类的main方法开始执行。

import org.apache.hadoop.conf.Configured;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class Main extends Configured implements Tool {
    @Override
    public int run(String[] args) throws Exception{
        // 默认配置套路
        Job job = Job.getInstance(getConf());
        job.setJobName("wordcount");
        job.setJarByClass(Main.class);
        // 设置输出的key/value类型
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        // 指定具体的Map和Reduce的类
        // 一个包内可能处理多种MapReduce
        // 因此可以写多个job，分别指定不同的Map和Reduce
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        // 设定输入输出的文件
        Path inputFilePath = new Path(args[0]);
        Path outputFilePath = new Path(args[1]);
        FileInputFormat.addInputPath(job, inputFilePath);
        FileOutputFormat.setOutputPath(job, outputFilePath);
        // 执行具体MapReduce
        return job.waitForCompletion(true) ? 0 : 1;
    }
    // 启动方法，按照套路来copy就行
    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new Main(), args);
        System.exit(exitCode);
    }
}

四、引入hadoop包

前面的类型完了，发现各种import的方法不存在，需要指定具体的hadoop的jar包。具体在IDEA中的操作如下，分别引入hadoop-2.7.0的common，mapreduce，yarn，hdfs，common/lib几个目录下的所有包

v2-c780ca13bb7ee95a9b3b6c0d8af695ee_720w.jpg

五、打包上传到docker中执行

打包设置如下

v2-8b90aee391d1ae8efd69c21e8a145b1f_720w.jpg

然后build Artifacts就可以打包了

v2-6cf90796f5e8744cc18eed083113ffc9_720w.jpg

将包传递到docker中，并执行

#上传到docker
docker cp Documents/git_code/wordcount/out/artifacts/wordcount_jar/wordcount.jar master:/usr/local
#进入docker
docker exec -it master bash
#编辑测试文件
cd /usr/local/
vi testFile.txt
#编辑测试文件内容为
Hello world
tes tFile
123 dd
ddd dd
3434s dfdsf df
242343 43252
sdfs df
#建立hadoop目录并上传文件
hadoop fs -mkdir -p /test/input
hadoop fs -put ./testFile.txt /test/input
#执行jar包，会自动生成output目录，以及目录下会生成文件
hadoop jar wordcount.jar /test/input/testFile.txt /test/output
#执行完成后，查看输出文件
hadoop fs -cat /test/output/*
#输出，统计处刚才testFile.txt的各个文字的计数
123 1
242343  1
3434s   1
43252   1
Hello   1
dd  2
ddd 1
df  2
dfdsf   1
sdfs    1
tFile   1
tes 1
world   1

MapReduce到这里就完事了，大家有什么不懂得可以在评论回复我，我来给大家详细解答。

谢各位的阅读，谢谢您动动手指点赞，万分感谢各位。另外以下是我之前写过的文章，感兴趣的可以点进去继续阅读。

历史文章