点一下关注吧！！！非常感谢！！持续更新！！！
🚀 AI 篇持续更新中！（长期更新）
目前 2025 年 06 月 05 日更新到：AI 炼丹日志-28 - Audiblez 将你的电子书 epub 转换为音频 mp3 做有声书，持续打造实用 AI 工具指南！📐🤖
💻 Java 篇正式开启！（300 篇）
目前 2025 年 06 月 11 日更新到：Java-42 深入浅出 Nginx - 缘起与发展 场景与配置快速上手 MyBatis 已完结，Spring 已完结，深入浅出助你打牢基础！
📊 大数据板块已完成多项干货更新（300 篇）：
包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！目前 2025 年 06 月 05 日更新到：大数据-278 Spark MLib - 基础介绍 机器学习算法 梯度提升树 GBDT 案例 详解
👉 点个关注，不迷路！后续还将持续更新更多大模型+数据智能+工程实战内容，敬请期待！
章节内容
上一节我们完成了：
背景介绍
这里是三台公网云服务器，每台 2C4G，搭建一个 Hadoop 的学习环境，供我学习。之前已经在 VM 虚拟机上搭建过一次，但是没留下笔记，这次趁着前几天薅羊毛的 3 台机器，赶紧尝试在公网上搭建体验一下。
2C4G 编号 h121
2C4G 编号 h122
2C2G 编号 h123
MapReduce 图片介绍
Hadoop 序列化
为什么是Hadoop实现的序列化，而不是使用Java自带的？
Mapper 规范
Reducer 规范
用户自定义 Reducer 类要继承 Hadoop 的 Reducer 类
Reducer 的输入数据类型对应 Mapper 的数据类型
Reducer 的业务逻辑在 reduce()方法中
Reduce()方法是对相同的 K 的一组 KV 对 调用执行一次
Driver 规范
创建提交 YARN 集群运行的 JOB 对象，其中封装了 MapReduce 程序运行所需要的相关参数：
输入数据路径
输出数据路径
Mapper
Reducer
也相当于一个 YRAN 集群的客户端，主要作用就是提交我们的 MapReduce 程序运行。
WordCount
需求介绍
下面是我做的一些操作，这里有一个 1.txt 文档，当中是一些文本内容。我们将对其做计算，统计出每个单词出现的频率。
root@hecs-393573:/opt/wzk# lswordcount.txtroot@hecs-393573:/opt/wzk# hadoop fs -get /wzk/test/1.txt ./root@hecs-393573:/opt/wzk# ls1.txt wordcount.txtroot@hecs-393573:/opt/wzk# hadoop fs -ls /Found 5 items
复制代码
操作输出的结果 如下
实现步骤
创建工程
导入 POM
编写 Mapper
编写 Reducer
编写 Driver
POM
<?xml version="1.0" encoding="UTF-8"?><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.0</modelVersion>
<groupId>org.example</groupId> <artifactId>hadoop-demo</artifactId> <version>1.0-SNAPSHOT</version>
<properties> <maven.compiler.source>8</maven.compiler.source> <maven.compiler.target>8</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties>
<dependencies> <!-- Hadoop Dependencies --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.9.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.9.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-core</artifactId> <version>2.9.0</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-mapreduce-client-common</artifactId> <version>2.9.0</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.8.2</version> </dependency> </dependencies>
<!--maven打包插件 --> <build> <plugins> <plugin> <artifactId>maven-compiler-plugin</artifactId> <version>2.3.2</version> <configuration> <source>1.8</source> <target>1.8</target> </configuration> </plugin> <plugin> <artifactId>maven-assembly-plugin</artifactId> <configuration> <descriptorRefs> <descriptorRef>jar-with-dependencies</descriptorRef> </descriptorRefs> </configuration> <executions> <execution> <id>make-assembly</id> <phase>package</phase> <goals> <goal>single</goal> </goals> </execution> </executions> </plugin> </plugins> </build></project>
复制代码
Mapper 编写
按照上述的规范，我们编写一个 Mapper 出来
package icu.wzk.demo02;
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 WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
Text k = new Text(); IntWritable v = new IntWritable(1);
@Override protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException { String line = value.toString(); String[] words = line.split(" "); for (String word : words) { k.set(word); context.write(k, v); } }}
复制代码
Reducer 编写
同样，按照之前文章中的说的规范，编写一个 Reducer 出来。
package icu.wzk.demo02;
import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
int sum; IntWritable v = new IntWritable();
@Override protected void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException { sum = 0; for (IntWritable count : values) { sum += count.get(); } v.set(sum); context.write(key, v); }}
复制代码
Driver 编写
package icu.wzk.demo02;
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.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.Job;
import java.io.IOException;
public class WordCountDriver {
public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { // String inputPath = args[0]; // String outputPath = args[1];
// === 测试 === String inputPath = "wc.txt"; String outputPath = "wc-out"; // =======
// 配置 Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); // 运行的类 job.setJarByClass(WordCountDriver.class); // Mapper job.setMapperClass(WordCountMapper.class); // Reducer job.setReducerClass(WordCountReducer.class); // Map Output Key Value job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 最终 Key Value job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 路劲参数 FileInputFormat.setInputPaths(job, new Path(inputPath)); FileOutputFormat.setOutputPath(job, new Path(outputPath)); // 等待结果 boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); }
}
复制代码
本地测试
在代码中，修改为本地的路径。
// === 测试 ===String inputPath = "wc.txt";String outputPath = "wc-out";// =======
复制代码
运行 Driver 代码之后，我们发现文件目录中生成了如下的结果
打开文本内容，我们可以看到如下的结果
评论