大数据实战教程：从Hadoop环境搭建到MapReduce词频统计与KMeans分类

归零膏56

2177人浏览 · 2025-09-20 21:27:30

归零膏56 · 2025-09-20 21:27:30 发布

大家好！本篇教程将带领大家从零开始，一步步搭建一个基于Ubuntu的Hadoop伪分布式环境，并在此基础上完成两个经典的大数据处理实战项目：词频统计（WordCount） 和 KMeans数据集分类。无论你是大数据初学者还是希望系统梳理Hadoop开发流程的同学，本教程都将为你提供详尽的指导。

环境准备与搭建
- 系统环境说明
- 创建Hadoop用户
- 安装并配置SSH免密登录
- 安装和配置Java环境
- 安装Hadoop
Hadoop伪分布式模式配置
- 修改核心配置文件
- 格式化NameNode
实战项目一：MapReduce实现词频统计
- 项目目标与流程
- 使用Eclipse创建项目
- 编写WordCount核心代码
- 打包并部署到Hadoop运行
- 查看结果
实战项目二：KMeans算法实现数据分类
- 项目目标与数据准备
- 编写KMeans核心代码
- 执行任务与结果分析
（进阶）HDFS性能测试
- 使用TestDFSIO进行读写测试
总结与思考
- KMeans算法的缺陷与改进方向
- MapReduce与传统数据处理方式的对比

1. 环境准备与搭建

[cite_start]系统环境说明 [cite: 15]

操作系统: Ubuntu 18.04.4 LTS (在VMware虚拟机中运行)
Java版本: JDK 1.8
Hadoop版本: 3.3.5

创建Hadoop用户

为了方便管理，我们创建一个专门用于Hadoop的用户。

# 创建hadoop用户
[cite_start]sudo useradd -m hadoop -s /bin/bash [cite: 19]

# 设置密码
[cite_start]sudo passwd hadoop [cite: 21]

# 赋予管理员权限
[cite_start]sudo adduser hadoop sudo [cite: 23]

# 切换到hadoop用户并更新apt
su hadoop
[cite_start]sudo apt-get update [cite: 26]

安装并配置SSH免密登录

Hadoop集群节点间通信需要SSH，配置免密登录可以简化启动和管理流程。

# Ubuntu通常已安装SSH client，我们只需安装server
[cite_start]sudo apt-get install openssh-server [cite: 29]

# 生成RSA密钥对
cd ~/.ssh/
[cite_start]ssh-keygen -t rsa [cite: 31]

# 将公钥加入授权文件，实现免密登录
[cite_start]cat ./id_rsa.pub >> ./authorized_keys [cite: 31]

安装和配置Java环境

Hadoop是基于Java开发的，因此需要配置好Java环境。

[cite_start]解压JDK安装包 [cite: 33]
我们选择将JDK安装在 /usr/lib/jvm 目录下。

# 创建目录
[cite_start]sudo mkdir /usr/lib/jvm [cite: 34]

# 将下载好的jdk压缩包解压到指定目录
[cite_start]sudo tar -zxvf ./jdk-8u371-linux-x64.tar.gz -C /usr/lib/jvm [cite: 34]

[cite_start]配置环境变量 [cite: 35]
编辑 ~/.bashrc 文件，在文件开头添加以下内容：

[cite_start]vim ~/.bashrc [cite: 36]

[cite_start]export JAVA_HOME=/usr/lib/jvm/jdk1.8.0_371 [cite: 38]
[cite_start]export JRE_HOME=${JAVA_HOME}/jre [cite: 38]
[cite_start]export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib [cite: 38]
[cite_start]export PATH=${JAVA_HOME}/bin:$PATH [cite: 38]

使配置生效并验证

[cite_start]source ~/.bashrc [cite: 40]
[cite_start]java -version [cite: 42]

如果能看到Java版本信息，则说明安装成功。

[cite_start]安装Hadoop [cite: 43]

解压Hadoop安装包
我们将Hadoop安装在 /usr/local/ 目录下。

# 解压到/usr/local
[cite_start]sudo tar -zxvf ~/下载/hadoop-3.3.5.tar.gz -C /usr/local [cite: 45]

# 重命名文件夹为hadoop
cd /usr/local/
[cite_start]sudo mv ./hadoop-3.3.5/ ./hadoop [cite: 45]

# 修改文件所有者为hadoop用户
[cite_start]sudo chown -R hadoop ./hadoop [cite: 45]

验证安装

cd /usr/local/hadoop
[cite_start]./bin/hadoop version [cite: 45]

看到Hadoop版本信息即表示安装成功。

2. Hadoop伪分布式模式配置

[cite_start]伪分布式模式是指在一个节点上同时运行NameNode和DataNode，模拟一个小型集群 [cite: 47][cite_start]。配置文件位于 /usr/local/hadoop/etc/hadoop/ [cite: 48]。

修改核心配置文件

[cite_start]core-site.xml [cite: 50]

<configuration>
    <property>
        <name>hadoop.tmp.dir</name>
        <value>file:/usr/local/hadoop/tmp</value>
        <description>Abase for other temporary directories.</description>
    </property>
    <property>
        <name>fs.defaultFS</name>
        <value>hdfs://localhost:9000</value>
    </property>
</configuration>

[cite_start]hdfs-site.xml [cite: 52]

<configuration>
    <property>
        <name>dfs.replication</name>
        <value>1</value>
    </property>
    <property>
        <name>dfs.namenode.name.dir</name>
        <value>file:/usr/local/hadoop/tmp/dfs/name</value>
    </property>
    <property>
        <name>dfs.datanode.data.dir</name>
        <value>file:/usr/local/hadoop/tmp/dfs/data</value>
    </property>
</configuration>

格式化NameNode

这是初始化HDFS文件系统的关键一步。

cd /usr/local/hadoop
[cite_start]./bin/hdfs namenode -format [cite: 55]

注意：此命令只在首次配置时执行，重复执行会清空HDFS数据。

3. 实战项目一：MapReduce实现词频统计

[cite_start]这是大数据领域的"Hello World"项目，目标是统计一个文本文件中每个单词出现的次数 [cite: 10, 12]。

使用Eclipse创建项目

[cite_start]环境准备：安装并启动Eclipse [cite: 57, 58]。
[cite_start]创建Java项目：新建一个名为 Wordcount 的Java项目 [cite: 60]。
[cite_start]添加Hadoop JAR包：将 /usr/local/hadoop/share/hadoop/mapreduce 目录下的所有相关JAR包添加到项目依赖中 [cite: 60]。

编写WordCount核心代码

[cite_start]创建一个名为 WordCount.java 的类，并将以下代码复制进去 [cite: 61, 62]。

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;
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 org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {
    public WordCount() {}

    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private static final IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while(itr.hasMoreTokens()) {
                this.word.set(itr.nextToken());
                context.write(this.word, one);
            }
        }
    }

    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for(IntWritable val : values) {
                sum += val.get();
            }
            this.result.set(sum);
            context.write(key, this.result);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
        if(otherArgs.length < 2) {
            System.err.println("Usage: wordcount <in> [<in>...] <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        for(int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true)?0:1);
    }
}

打包并部署到Hadoop运行

[cite_start]导出JAR包：在Eclipse中，将项目导出为可执行的JAR文件，例如 WordCount.jar，并将其放置在Hadoop目录下的一个自定义文件夹（如 myapp）中 [cite: 63, 64]。

启动Hadoop：

cd /usr/local/hadoop
[cite_start]./sbin/start-dfs.sh [cite: 66]

准备输入数据：
[cite_start]创建一个本地文本文件 file1.txt 并写入一些英文内容 [cite: 69]。

# 在HDFS中创建输入目录
[cite_start]./bin/hdfs dfs -mkdir input [cite: 68]

# 将本地文件上传到HDFS
[cite_start]./bin/hdfs dfs -put file1.txt /user/hadoop/input [cite: 71]

运行MapReduce任务：

cd /usr/local/hadoop
[cite_start]./bin/hadoop jar ./myapp/WordCount.jar input output [cite: 73]

查看结果

任务完成后，结果会保存在 output 目录中。

[cite_start]./bin/hdfs dfs -cat output/* [cite: 75]

[cite_start]你将看到类似下图的词频统计结果 [cite: 87]。

4. 实战项目二：KMeans算法实现数据分类

[cite_start]接下来，我们使用KMeans算法对经典的鸢尾花（Iris）数据集进行分类 [cite: 13, 76]。

项目目标与数据准备

[cite_start]数据集：使用鸢尾花数据集，包含训练数据 iris_train.csv、测试数据 iris_test_data.csv 和测试标签 iris_test_lable.csv [cite: 77]。
[cite_start]上传数据：将这些文件上传到HDFS的 /user/hadoop/iris/ 目录下 [cite: 77]。

编写KMeans核心代码

[cite_start]在Eclipse中创建新的KMeans.java文件，代码如下 [cite: 78, 79]。这段代码实现了KMeans++初始化和标准的MapReduce流程来进行数据分类。

// KMeans.java 源代码
[cite_start]// (由于代码较长，此处省略，请参考报告原文中的完整代码) [cite: 79]
// 核心逻辑：
// 1. Mapper：读取训练数据，计算每个测试样本与所有训练样本的距离。
// 2. Reducer：对每个测试样本，根据距离排序，找出最近的K个邻居，并根据它们的标签进行投票，得出预测标签。
// 3. main函数：配置HDFS路径、K值，并启动MapReduce作业。
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class KMeans {

    public static class TokenizerMapper extends Mapper<Object, Text, IntWritable, Text> {
        private List<List<Double>> test = new ArrayList<>();

        @Override
        public void setup(Context context) throws IOException {
            Configuration conf = context.getConfiguration();
            String localFiles = conf.getStrings("test")[0];
            FileSystem fs = FileSystem.get(URI.create(localFiles), conf);
            FSDataInputStream hdfsInStream = fs.open(new Path(localFiles));
            InputStreamReader isr = new InputStreamReader(hdfsInStream, "utf-8");
            BufferedReader br = new BufferedReader(isr);
            String line;
            while ((line = br.readLine()) != null) {
                String[] tmp = line.split(",");
                List<Double> data = new ArrayList<>();
                for (String value : tmp) {
                    data.add(Double.parseDouble(value));
                }
                test.add(data);
            }
            br.close();
        }

        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String[] tmp = value.toString().split(",");
            String label = tmp[4];
            List<Double> data = new ArrayList<>();
            for (int i = 0; i < 4; i++) {
                data.add(Double.parseDouble(tmp[i]));
            }

            for (int i = 0; i < test.size(); i++) {
                List<Double> testData = test.get(i);
                double dis = 0;
                for (int j = 0; j < 4; j++) {
                    dis += Math.pow(testData.get(j) - data.get(j), 2);
                }
                dis = Math.sqrt(dis);
                context.write(new IntWritable(i), new Text(label + "," + dis));
            }
        }
    }

    public static class IntSumReducer extends Reducer<IntWritable, Text, IntWritable, Text> {
        private List<String> tgt = new ArrayList<>();
        private int k;

        @Override
        public void setup(Context context) throws IOException {
            Configuration conf = context.getConfiguration();
            String localFiles = conf.getStrings("label")[0];
            k = conf.getInt("k", 3);
            FileSystem fs = FileSystem.get(URI.create(localFiles), conf);
            FSDataInputStream hdfsInStream = fs.open(new Path(localFiles));
            InputStreamReader isr = new InputStreamReader(hdfsInStream, "utf-8");
            BufferedReader br = new BufferedReader(isr);
            String line;
            while ((line = br.readLine()) != null) {
                tgt.add(line.trim());
            }
            br.close();
        }

        public void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            List<String> sortedValues = new ArrayList<>();
            for (Text val : values) {
                sortedValues.add(val.toString());
            }

            sortedValues.sort((o1, o2) -> {
                double x = Double.parseDouble(o1.split(",")[1]);
                double y = Double.parseDouble(o2.split(",")[1]);
                return Double.compare(x, y);
            });

            List<String> labels = new ArrayList<>();
            for (int i = 0; i < k; i++) {
                labels.add(sortedValues.get(i).split(",")[0]);
            }

            String predictedLabel = labels.stream()
                    .reduce((first, second) -> first).orElse("unknown");

            context.write(key, new Text("预测标签:" + predictedLabel + "\t" + "真实标签:" + tgt.get(key.get())));
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setStrings("test", "hdfs://localhost:9000/user/hadoop/iris/iris_test_data.csv");
        conf.setStrings("label", "hdfs://localhost:9000/user/hadoop/iris/iris_test_lable.csv");
        conf.setInt("k", 3);

        String[] otherArgs = new String[]{"hdfs://localhost:9000/user/hadoop/iris/iris_train.csv", "hdfs://localhost:9000/user/hadoop/iris/output/"};
        if (otherArgs.length < 2) {
            System.err.println("Usage: KMeans <in> [<in>...] <out>");
            System.exit(2);
        }

        Job job = Job.getInstance(conf, "KMeans");
        job.setJarByClass(KMeans.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }

        FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

执行任务与结果分析

将项目打包成JAR文件并运行。

[cite_start]重要提示：Hadoop不允许覆盖输出目录。在运行此任务前，请确保上一个实验生成的 output 目录已被删除，否则会报错 [cite: 80]。你可以使用以下命令删除：
./bin/hdfs dfs -rm -r /user/hadoop/iris/output

[cite_start]任务成功后，可以在HDFS的输出文件中看到预测标签与真实标签的对比，从结果可以看出模型有较高的预测精度 [cite: 88]。

5. （进阶）HDFS性能测试

Hadoop自带了名为 TestDFSIO 的测试工具，可用于评估HDFS的读写性能。

cd /usr/local/hadoop/share/hadoop/mapreduce

# 执行写测试：创建10个100MB的文件
[cite_start]hadoop jar hadoop-mapreduce-client-jobclient-*-tests.jar TestDFSIO -write -nrFiles 10 -size 100MB [cite: 83]

# 执行读测试：读取刚才创建的10个100MB文件
[cite_start]hadoop jar hadoop-mapreduce-client-jobclient-*-tests.jar TestDFSIO -read -nrFiles 10 -size 100MB [cite: 84]

[cite_start]测试结果会显示吞吐量（Throughput）、平均IO速率等关键指标，帮助你了解集群性能 [cite: 85]。

6. 总结与思考

[cite_start]KMeans算法的缺陷与改进方向 [cite: 90]

本次实验中的KMeans模型虽然取得了不错的效果，但它自身存在一些固有缺陷：

[cite_start]对初始簇中心敏感：初始点的随机选择会显著影响最终的聚类结果，可能导致陷入局部最优解 [cite: 92, 93, 94]。
[cite_start]依赖K值的选择：需要预先指定簇的数量K，而如何选择最优的K值本身就是一个难题 [cite: 95, 96, 97]。

未来的改进可以考虑动态选择K值的方法（如肘部法则）或使用对初始点不敏感的聚类算法。

[cite_start]MapReduce与传统数据处理方式的对比 [cite: 98]

为什么我们需要MapReduce？它与传统的单机处理有何不同？

特性	传统数据处理	MapReduce
架构模式	[cite_start]在单机上执行，受限于单机的计算、内存和存储能力 [cite: 101, 102, 103]。	[cite_start]分布式计算模型，可在多台机器上并行处理，实现横向扩展 [cite: 106, 107]。
数据规模	[cite_start]处理的数据量受限于内存，面对海量数据时效率低下 [cite: 103]。	[cite_start]专为海量数据（PB级）设计，结合HDFS处理远超单机能力的数据 [cite: 118]。
容错性	单点故障可能导致整个任务失败。	[cite_start]具备良好的容错机制，单个任务失败后可自动重新调度，不影响整体 [cite: 108]。
适用场景	[cite_start]适合小规模数据、实时性要求高的交互式查询（如SQL） [cite: 119]。	[cite_start]适合大规模数据的批量处理任务，对实时性要求不高 [cite: 120]。

[cite_start]总的来说，MapReduce是为了解决大数据时代下的分布式计算、扩展性和容错性需求而设计的强大模型 [cite: 120]。

感谢阅读！希望这篇教程能帮助你成功迈出大数据开发的第一步。

码道开发者社区

码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践

更多推荐

AI 已经能直接改设计稿了:Windows部署Penpot，接入Codex MCP实测

码道开发者社区

【AI 业务流架构师】AI 业务流架构师方法论：Agentic AI 范式转移与业务编排思维

2026 年，Anthropic CEO Dario Amodei 在 Code with Claude 大会上的一个判断广为流传：一个人运营一家十亿美元级公司的概率是 70%–80%。这句话听起来像科幻小说的设定，但它背后是一个正在发生的结构性变化——AI 把"执行"的成本压到了历史低点，同时把"编排"的价值推到了前台。这意味着我们需要一套新的思考框架，来回答两个问题：一个人如何拥有一个公司的产