实训云上搭建大数据集群

• 本次实训在华为云实训云上从零搭建三节点 Hadoop 集群，再用 MapReduce、Hive、Spark 三套技术栈实现同一个词频统计任务，完整走通「基础设施 → 网络安全 → 主机接入 → 集群配置 → 应用开发」全流程。

1. 登录实训云

1.1 实训云网址

• 网址：http://192.168.192.40
  

1.2 登录实训云

• 输入用户名和密码
  

• 登录成功后显示首页
  

2. 创建网络

2.1 网络概述

• 网络名为huawei，子网为huawei_subnet，网络地址为192.168.1.0/24，分配地址池192.168.1.2,192.168.1.254，DNS设置为114.114.114.114。此网络结构清晰，地址配置明确，具备基础的网络配置要素，可满足一般网络应用场景下的设备接入与通信需求。

2.2 创建步骤

• 显示【网络】页面
  

• 单击【创建网络】按钮，在弹出的对话框里设置网络基本信息
  

• 单击【确定】按钮，成功创建网络huawei
  

3. 创建路由器

3.1 路由器名称

• 路由器名称：huawei_router

3.1 创建路由器

• 显示【路由器】页面
  

• 单击【创建路由器】按钮，在弹出的对话框里设置路由器基本信息
  

• 单击【确定】按钮，成功创建路由器
  

4. 连接子网

• 显示【路由器】页面
  

• 单击【更多】右边的箭头弹出下拉菜单
  

• 单击【连接子网】菜单选项，连接网络huawei与子网huawei_subnet
  

• 单击【确定】按钮，提示连接子网成功
  

5. 创建虚拟网卡

5.1 创建原因

• 创建虚拟网卡是为了给云主机分配固定IP地址。云主机在网络中运行，若需稳定的网络标识和连接，固定IP不可或缺。虚拟网卡充当云主机与网络间的桥梁，通过创建它并绑定固定IP，可确保云主机在网络中的唯一性和稳定性，便于管理与访问。

5.2 查看端口

• 显示【端口】页面
  

5.3 创建虚拟网卡

• 单击【创建虚拟网卡】按钮，弹出对话框设置虚拟网卡基本信息
  

• 单击【确定】按钮，成功创建虚拟网卡port101
  

• 同理创建虚拟网卡port102和port103
  

6. 管理安全组规则

6.1 为什么要管理安全组规则

• 管理安全组规则至关重要，因为后续使用FinalShell工具软件通过SSH连接云主机时，需设置规则放开ICMP协议和TCP协议入口的全部端口。这样能确保连接的顺利进行，同时保障网络安全，在满足连接需求的前提下，有效控制网络访问权限，防止潜在的安全威胁。

6.2 在默认安全组里创建规则

• 显示【安全组】页面
  

• 单击【创建规则】按钮，弹出对话框
  

• 设置所有TCP协议
  

• 单击【确定】按钮，提示“创建规则成功”
  

• 单击【创建规则】按钮，弹出对话框，设置所有ICMP协议
  

• 单击【确定】按钮，提示“创建规则成功”
  

7. 创建云主机

7.1 创建云主机bigdata1

• 显示【云主机】页面
  

• 单击【创建云主机】按钮，在弹出的对话框里设置云主机基本信息
  

• 单击【下一步：网络配置】按钮，在弹出的对话框里选择port101端口，勾选default安全组
  

• 单击【下一步：系统配置】按钮，在弹出的对话框里设置名称与登录密码
  

• 单击【下一步：确认配置】按钮
  

• 单击【确定】按钮，成功创建云主机bigdata1，有固定IP地址192.168.1.101
  

• 内网IP有两个，我们要解绑192.168.1.46
  

• 云主机bigdata1内网IP：192.168.1.101
  

7.2 创建云主机bigdata2

• 同理创建云主机bigdata2，有固定IP地址192.168.1.102
  

7.3 创建云主机bigdata3

• 同理创建云主机bigdata3，有固定IP地址192.168.1.103
  

8. 绑定浮动IP地址

8.1 为何要绑定

• 绑定浮动IP地址是实现FinalShell远程连接云主机的关键步骤。浮动IP地址具有灵活性，可与云主机动态绑定，即便云主机发生故障迁移等情况，IP地址仍可保持不变，从而确保FinalShell能够稳定地远程连接到云主机，保障远程操作的连续性和可靠性。

8.2 申请浮动IP地址

• 显示【浮动IP】页面
  

• 单击【申请IP】按钮，在弹出的对话框里设置网络信息与批量申请数量
  

• 单击【确定】按钮，成功申请了三个浮动IP地址
  

8.3 绑定浮动IP地址

• 显示【云主机】页面

• 给云主机bigdata1绑定浮动IP地址，【更多】下拉菜单的子菜单【关联资源】，【关联资源】的子菜单【绑定浮动IP】
  

• 单击【绑定浮动IP】菜单项，将云主机IP与浮动IP地址进行绑定
  

• 单击【确定】按钮，成功给云主机bigdata1绑定浮动IP地址
  

• 同理，给云主机bigdata2和bigdata3绑定相应的浮动IP地址
  

9. 远程连接云主机

9.1 启动FinalShell

• 启动FinalShell，目前尚未创建任何远程连接
  

9.2 远程连接云主机bigdata1

• 弹出【连接管理器】
  

• 新建SSH连接
  

• 单击【SSH连接(Linux)】，在弹出的对话框里设置常规和认证信息，注意密码是123456
  

• 单击【确定】按钮，成功创建连接bigdata1
  

• 单击bigdata1连接，弹出【安全警告】对话框
  

• 单击【接受并保存】按钮
  

9.3 远程连接云主机bigdata2

• 同理，远程连接云主机bigdata2
  

9.4 远程连接云主机bigdata3

• 同理，远程连接云主机bigdata3
  

10. 修改本地的IP与主机名映射文件

• 在文件C:\Windows\System32\drivers\etc\hosts里添加三条映射
  

  192.168.224.36bigdata1192.168.227.230bigdata2192.168.228.14bigdata3

11. 本机Ping云主机

11.1 按IP地址来Ping

• 执行命令：ping 192.168.224.36
  

11.2 按主机名来Ping

• 执行命令：ping bigdata1
  

12. 修改配置文件

12.1 修改hdfs-site.xml文件

• 执行命令：cd $HADOOP_HOME/etc/hadoop
  
• 执行命令：vim hdfs-site.xml，添加红框内属性，禁用所有权限检查
  
• 执行命令：scp hdfs-site.xml root@bigdata2:$HADOOP_HOME/etc/hadoop
  
• 执行命令：scp hdfs-site.xml root@bigdata3:$HADOOP_HOME/etc/hadoop
  

12.2 修改log4j.properties文件

• 执行命令：vim log4j.properties，添加一句：log4j.logger.org.apache.hadoop=ERROR
  
• 执行命令：scp log4j.properties root@bigdata2:$HADOOP_HOME/etc/hadoop
  
• 执行命令：scp log4j.properties root@bigdata3:$HADOOP_HOME/etc/hadoop
  

13. 启动大数据集群服务

• 在云主机bigdata1上，执行命令：allstart.sh
  

• 执行命令：jps
  

• 查看Hadoop的Web UI页面 -http://bigdata1:9870
  

• 查看HDFS文件系统
  

14. 完成词频统计任务

14.1 准备数据文件

• 执行命令：vim words.txt
  

  hello world hello hadoop hello hive hello spark hello a big data world hello I love big data hello I learn big data

• 执行命令：hdfs dfs -mkdir -p /wordcount/input
  

• 执行命令：hdfs dfs -put words.txt /wordcount/input
  

14.2 利用MR实现词频统计

14.2.1 创建Maven项目

• 设置项目基本信息
  
• 单击【Create】按钮，生成项目基本骨架
  

14.2.2 添加hadoop依赖

• 在pom.xml里添加hadoop和junit依赖
  
• 刷新项目依赖
  

14.2.3 创建日志属性文件

• 在resources里创建log4j.properties文件
  

  log4j.rootLogger=ERROR, stdout, logfilelog4j.appender.stdout=org.apache.log4j.ConsoleAppenderlog4j.appender.stdout.layout=org.apache.log4j.PatternLayoutlog4j.appender.stdout.layout.ConversionPattern=%d %p[%c]- %m%nlog4j.appender.logfile=org.apache.log4j.FileAppenderlog4j.appender.logfile.File=target/wordcount.loglog4j.appender.logfile.layout=org.apache.log4j.PatternLayoutlog4j.appender.logfile.layout.ConversionPattern=%d %p[%c]- %m%n

14.2.4 创建词频统计映射器类

• 创建net.huawei.mr包
  

• 在net.huawei.mr包里创建WordCountMapper类
  

  packagenet.huawei.mr;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;importjava.io.IOException;/** * 功能：词频统计映射器 * 作者：华卫 * 日期：2025年12月03日 */publicclassWordCountMapperextendsMapper<LongWritable,Text,Text,IntWritable>{@Overrideprotectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{// 获取行文本Stringline=value.toString();// 按空格拆分成单词数组String[]words=line.split(" ");// 遍历单词数组，生成输出键值对for(Stringword:words){context.write(newText(word),newIntWritable(1));}}}

• 代码说明：该代码定义了一个 Hadoop MapReduce 的 Mapper 类，用于词频统计。它将输入的每行文本按空格拆分为单词，对每个单词输出键值对（单词, 1），供后续 Reduce 阶段累加计数。继承自Mapper<LongWritable, Text, Text, IntWritable>，符合 WordCount 任务规范。

14.2.5 创建词频统计归并器类

• 在net.huawei.mr包里创建WordCountReducer类
  

  packagenet.huawei.mr;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.NullWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Reducer;importjava.io.IOException;/** * 功能：词频统计归并器类 * 作者：华卫 * 日期：2025年12月03日 */publicclassWordCountReducerextendsReducer<Text,IntWritable,Text,NullWritable>{@Overrideprotectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{// 定义键（单词）出现次数intcount=0;// 遍历值迭代器，累加次数for(IntWritablevalue:values){count=count+value.get();}// 生成新的输出键StringnewKey="("+key.toString()+","+count+")";// 输出新的键值对(newKey, null)context.write(newText(newKey),NullWritable.get());}}

• 代码说明：该代码实现 Hadoop MapReduce 的 Reducer 类，用于词频统计。它接收相同单词的所有计数（值为1的 IntWritable），累加得到总频次，并将结果格式化为 “(单词,次数)” 字符串作为新键输出，值设为 NullWritable，适用于仅需输出统计结果的场景。

14.2.6 创建词频统计驱动器类

• 在net.huawei.mr包里创建WordCountDriver类
  

  packagenet.huawei.mr;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.FSDataInputStream;importorg.apache.hadoop.fs.FileStatus;importorg.apache.hadoop.fs.FileSystem;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IOUtils;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.NullWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;importjava.net.URI;/** * 功能：词频统计驱动器 * 作者：华卫 * 日期：2025年12月03日 */publicclassWordCountDriver{publicstaticvoidmain(String[]args)throwsException{// 创建配置对象Configurationconf=newConfiguration();// 设置客户端使用数据节点主机名属性conf.set("dfs.client.use.datanode.hostname","true");// 获取工作实例Jobjob=Job.getInstance(conf);// 设置作业启动类job.setJarByClass(WordCountDriver.class);// 设置Mapper类job.setMapperClass(WordCountMapper.class);// 设置map任务输出键类型job.setMapOutputKeyClass(Text.class);// 设置map任务输出值类型job.setMapOutputValueClass(IntWritable.class);// 设置Reducer类job.setReducerClass(WordCountReducer.class);// 设置reduce任务输出键类型job.setOutputKeyClass(Text.class);// 设置reduce任务输出值类型job.setOutputValueClass(NullWritable.class);// 定义uri字符串（分布式文件系统前缀）Stringuri="hdfs://bigdata1:9000";// 创建输入路径（可以指向目录，也可以指向文件）PathinputPath=newPath(uri+"/wordcount/input");// 创建输出路径PathoutputPath=newPath(uri+"/wordcount/output");// 获取文件系统FileSystemfs=FileSystem.get(newURI(uri),conf);// 删除输出路径（第2个参数设置是否递归删除）fs.delete(outputPath,true);// 给作业添加输入路径（允许多个）FileInputFormat.addInputPath(job,inputPath);// 给作业设置输出路径（只能一个）FileOutputFormat.setOutputPath(job,outputPath);// 等待作业完成job.waitForCompletion(true);// 输出统计结果System.out.println("======统计结果======");// 获取输出路径的文件状态数组FileStatus[]fileStatuses=fs.listStatus(outputPath);// 遍历文件状态数组for(inti=1;i<fileStatuses.length;i++){// 输出结果文件路径System.out.println(fileStatuses[i].getPath());// 获取文件系统字节输入流FSDataInputStreamin=fs.open(fileStatuses[i].getPath());// 将结果文件显示在控制台IOUtils.copyBytes(in,System.out,4096,false);}}}

• 代码说明：该驱动类配置并提交 Hadoop 词频统计作业：设置 Mapper 和 Reducer 类及其输入输出类型，指定 HDFS 输入/输出路径，自动删除已有输出目录，作业完成后读取并打印结果文件内容到控制台。

14.2.7 运行程序，查看结果

• 运行词频统计驱动器类
  

14.3 利用Hive实现词频统计

• 执行命令：hive，进入Hive客户端
  

14.3.1 基于HDFS文件创建外部表

• 执行语句：create external table t_word(line string) location '/wordcount/input';
  

14.3.2 展开每行单词

• 执行语句：select explode(split(line, ' ')) as word from t_word;
  

14.3.3 基于查询结果创建视图

• 执行语句：create view v_word as select explode(split(line, ' ')) as word from t_word;

14.3.4 基于视图分组统计

• 不排序统计，执行语句：select word, count(*) from v_word group by word;
  
• 按词频排序统计，执行语句：select word, count(*) as count from v_word group by word order by count desc;
  
  

14.4 利用Spark实现词频统计

• 执行命令：spark-shell
  

14.4.1 利用Spark RDD实现词频统计

• 执行代码

  sc.textFile("hdfs://bigdata1:9000/wordcount/input/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).sortBy(_._2,false).collect().foreach(println)

14.4.2 利用Spark SQL实现词频统计

1. 读取文本文件生成数据帧

   • 执行命令：val df = spark.read.text("hdfs://bigdata1:9000/wordcount/input/words.txt")
     

   • 执行命令：df.show()
     

2. 基于数据帧生成临时视图

   • 执行命令：df.createOrReplaceTempView("t_value")
     

3. 对临时视图进行词频统计

   • 执行代码

     valwc=spark.sql(""" select word, count(*) as count from (select explode(split(value, " ")) as word from t_value) group by word order by count desc """)

   • 执行命令：wc.show()
     

15. 关闭大数据集群服务

• 在云主机bigdata1上，执行命令：allstop.sh
  

16. 实战小结

• 通过本次实战，我们完整地搭建并运行了一个基于Hadoop、Hive和Spark的大数据集群环境。从网络和云主机的创建、配置，到集群服务的启动与管理，再到利用MapReduce、Hive和Spark实现词频统计任务，全面掌握了大数据平台的搭建与应用开发流程。在实战中，我们成功实现了数据的分布式存储与处理，验证了不同技术框架在数据处理中的优势和适用场景。通过FinalShell远程连接和配置文件修改，进一步增强了对集群管理的理解。整个过程不仅提升了技术操作能力，还加深了对大数据生态系统各组件协同工作的认识，为后续更复杂的大数据项目积累了宝贵经验。