MapReduce编程模型——在idea里面邂逅CDH MapReduce

🏠MapReduce

MapReduce是一个分布式运算程序的编程框架，它是hadoop的重要组成部分，其主要负责分布式计算。MapReduce具有高容错性的优点，适合海量数据的离线处理。

MapReduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个Hadoop集群上。编写一个完整的MapReduce任务只需要三步：

1、编写mapper阶段的逻辑代码
2、编写reducer阶段的逻辑代码
3、将mapper阶段和reducer阶段组合，连同一些配置，组织成主任务，并提交任务。
1
2
3

更为复杂的任务可以多次重复以上三步。

本文在IDEA中编写基于CDH的MapReduce任务并操作CDH平台HDFS中的文件。
MapReduce的原理有很多可以参考的文章，这里只给出实操的部分。

🏠案例

本文要完成的目标是统计HDFS上文本文件中单词的个数，包括中文，单词的分隔符采用空格。HDFS用户采用匿名用户：anonymous，文件输入路径是：/user/anonymous/mrtest/one/input，输出路径是：/user/anonymous/mrtest/one/output。用户和路径读者均可自行修改。
程序处理流程大致如下：
在这里插入图片描述

👉环境

大数据平台：集群 CDH 6.2.3
本地开发环境：Windows10

👉部署本地开发环境

下载winutils-master并解压，解压后，目录内容如下：
在这里插入图片描述
将该目录中的 hadoop-3.0.0 文件夹路径设置为环境变量 HADOOP_HOME，将 hadoop-3.0.0\bin 中的 hadoop.dll 复制到 C:\Windows\System32 目录下。

👉编写代码

1、在IDEA中新建maven项目
2、登录CDH管理页面，下载HDFS的配置文件，如下：
在这里插入图片描述
文件下载到本地后，解压，将里面的 core-site.xml 和 hdfs-site.xml 两个文件复制到项目的resources目录下：

3、编写代码
项目的pom文件如下：


<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0modelVersion>

    <groupId>org.examplegroupId>
    <artifactId>hadooptestartifactId>
    <version>1.0-SNAPSHOTversion>

    <properties>
        <hadoop_version>3.0.0-cdh6.3.2hadoop_version>
    properties>

    <repositories>
        <repository>
            <id>clouderaid>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/url>
            <releases>
                <enabled>trueenabled>
            releases>
            <snapshots>
                <enabled>falseenabled>
            snapshots>
        repository>
    repositories>


    <dependencies>
        
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-commonartifactId>
            <version>${hadoop_version}version>
        dependency>
        
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-hdfsartifactId>
            <version>${hadoop_version}version>
        dependency>
        
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-mapreduce-client-coreartifactId>
            <version>${hadoop_version}version>
        dependency>
        
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-clientartifactId>
            <version>${hadoop_version}version>
        dependency>
        
        <dependency>
            <groupId>org.apache.hadoopgroupId>
            <artifactId>hadoop-yarn-apiartifactId>
            <version>${hadoop_version}version>
        dependency>
    dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.pluginsgroupId>
                <artifactId>maven-compiler-pluginartifactId>
                <version>3.8.1version>
                <configuration>
                    <source>1.8source>
                    <target>1.8target>
                    <encoding>UTF-8encoding>
                configuration>
            plugin>
        plugins>
    build>
project>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

日志输出配置如下：

log4j.rootLogger=INFO,console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n

1
2
3
4
5
6

Mapper类如下：

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;

public class MyMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        // 分词
        String[] words = value.toString().split(" ");
        Text text = new Text();
        IntWritable cnt = new IntWritable(1);
        for (String word : words) {
            text.set(word);
            context.write(text, cnt);
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

Reducer类如下

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

import java.io.IOException;

public class MyReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    private IntWritable result = new IntWritable();

    @Override
    protected 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);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

主程序如下：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class App {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 设置用户
        System.setProperty("HADOOP_USER_NAME", "anonymous");
        // 任务配置
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "my wordcount");
        job.setMapperClass(MyMapper.class);
        job.setReducerClass(MyReducer.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        job.setJarByClass(App.class);
        // 输入输出路径
        Path inputPath = new Path("mrtest/one/input");
        Path outputPath = new Path("mrtest/one/output");
        // 输出目录若存在，则进行删除
        FileSystem fileSystem = FileSystem.get(conf);
        if(fileSystem.exists(outputPath)){
            fileSystem.delete(outputPath, true);
        }
        // 给任务设置输入输出路径
        FileInputFormat.addInputPath(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);
        // 提交任务
        boolean flag = job.waitForCompletion(true);
        if (flag){
            System.out.println("程序运行结束！！");
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

作者这水平有限，有不足之处欢迎留言指正！！

相关阅读:
机器学习3判断机器算法的性能
 关于#matlab#的问题：图1是一个污染物动态演变模型
 Java成员变量与成员方法简介说明
 K8S自建LoadBalancer
【python】根据自定义曲线函数进行拟合
 2022 RoboCom 世界机器人开发者大赛-本科组（省赛）-RC-u5 树与二分图
 分布式文件存储系统FastDFS[2]-上传和下载文件工具类
 python创建空列表
 四种常用的自动化测试框架
 MOCK远程API调用的简单实现
原文地址：https://blog.csdn.net/weixin_37522117/article/details/127653994