[云] 大数据分析栈（Big Data Analytics Stack）+ Apache Hadoop分布式文件系统（HDFS）+Apache Spark

任务概述

本次作业旨在帮助你理解大数据分析栈（Big Data Analytics Stack）的工作原理，并通过实际操作加深认识。你将搭建Apache Hadoop分布式文件系统（HDFS）作为底层文件系统，并将Apache Spark作为执行引擎来运行应用程序。最终，你需要撰写一份简短报告，记录你的观察和收获。

作业分部分解与解析

Part 0: 环境设置

AWS Academy环境：作业要求在AWS Academy（Amazon Web Services培训环境）中进行。你需要创建两台EC2虚拟机实例，实例类型为 t3.large。

解析：EC2是AWS提供的虚拟服务器，适合测试和开发。t3.large类型的实例具有中等性能，适合本次实验中的小型分布式集群。
创建与命名EC2实例：创建两台相同类型的EC2实例，然后重命名实例名称（例如命名为vm1和vm2）以方便区分和访问。

解析：将实例命名为vm1和vm2有助于在配置时区分这两台虚拟机，便于操作。比如，知道自己是在vm1上搭建HDFS的主节点还是在vm2上搭建辅助节点。
搭建两节点集群：在vm1和vm2上部署两节点的HDFS集群和两节点的Spark集群。

解析：这要求你在两台虚拟机上分别安装并配置HDFS和Spark，使它们组成一个简单的分布式集群。在HDFS中，文件会被分片并分布到多台节点上，而Spark可以并行处理数据，从而更高效地完成任务。
安装Java：Hadoop HDFS和Spark都需要Java运行时环境，因此第一步是在每台虚拟机上安装Java。由于Ubuntu 22.04 LTS预装了Python 3.10.6，你无需手动安装Python。

关联知识点：
- Hadoop HDFS：一个分布式文件系统，能够跨多台机器保存数据分片，支持高可用性和容错。它是许多大数据处理系统的核心。
- Spark：一个基于内存的分布式计算框架，适合处理大规模数据。Spark的并行处理能力使它比传统的MapReduce更快。
- EC2与分布式集群：在云环境（如AWS）中搭建分布式集群，通过创建多个虚拟机，可以模拟真实的大数据处理场景。

指导与解析

任务：免密SSH设置与Java安装

1. 安装Java

首先，需在每个虚拟机（EC2实例）上更新系统并安装Java 8，这是Hadoop和Spark的依赖。

sudo apt update
sudo apt install -y openjdk-8-jdk

解析：
sudo apt update更新Ubuntu软件源，使系统获取到最新的软件列表。sudo apt install -y openjdk-8-jdk命令直接安装Java 8的开发工具包（JDK）。由于Hadoop和Spark依赖Java，这步为后续配置HDFS和Spark集群做好准备。

2. 设置免密SSH访问

在主节点（vm1）上执行以下命令生成密钥对：

ssh-keygen -t rsa

解析：
ssh-keygen -t rsa生成一个RSA加密算法的公私钥对。生成的私钥保存在~/.ssh/id_rsa文件中，公钥则在~/.ssh/id_rsa.pub中。生成密钥对后，可以通过公钥让vm1在无需密码的情况下访问其他节点。

3. 将公钥添加到`authorized_keys`文件

步骤：

获取公钥内容，使用以下命令查看id_rsa.pub文件内容：
```
cat ~/.ssh/id_rsa.pub
```
将该输出内容复制到~/.ssh/authorized_keys文件中，注意不要覆盖已有内容。

在vm1和vm2上操作：

在vm1上，将内容添加到~/.ssh/authorized_keys文件中，这样vm1可以访问自己。
在vm2上，打开~/.ssh/authorized_keys文件并将vm1的公钥内容粘贴进去，使vm1能够访问vm2。
不要添加空行：将公钥粘贴到authorized_keys时，避免额外的换行或空行，否则可能会导致连接失败。
不要覆盖已有内容：确保不删除authorized_keys中的其他密钥行，否则可能失去对实例的访问权限。

笔记：

这些步骤的核心是让两台虚拟机（vm1和vm2）可以在不输入密码的情况下相互通信，尤其是主节点（vm1）要能够直接控制和访问从节点（vm2）。这种配置对于自动化管理分布式系统（如HDFS和Spark集群）至关重要。

步骤分解与详细解释

1. 安装Java

命令：sudo apt update 和 sudo apt install -y openjdk-8-jdk
作用：更新系统并安装Java开发工具包（JDK），因为Hadoop和Spark是基于Java开发的，它们运行时需要Java环境支持。
通俗解释：就像安装了操作软件所需要的“工具箱”，这样Hadoop和Spark才能顺利启动和运行。

2. 生成SSH密钥对

命令：ssh-keygen -t rsa
作用：生成一对“钥匙”——一个私钥和一个公钥。私钥相当于钥匙的“芯”，仅保留在自己机器上，不可泄露；公钥相当于钥匙的“外壳”，可以提供给其他机器用来识别自己。
通俗解释：好比是给主节点（vm1）配备了一把钥匙对，它可以用自己的“芯”来解锁任何允许它访问的“锁”（即把公钥给其他机器）。这个操作只在主节点vm1上做。

3. 设置免密访问

命令：通过cat ~/.ssh/id_rsa.pub查看生成的公钥，并将其复制到~/.ssh/authorized_keys文件中。
作用：将公钥内容粘贴到authorized_keys文件里，授权对应密钥的用户（即vm1）可以免密访问。
操作步骤：
1. 在vm1上获取生成的公钥内容。
2. 将公钥内容粘贴到~/.ssh/authorized_keys文件中，在vm1和vm2都粘贴一遍。
通俗解释：
- 这一过程就像是将vm1的钥匙交给vm2，这样以后vm1访问vm2时无需“敲门”（输入密码）就能进门。同理，把公钥粘贴到vm1自己的authorized_keys中，可以让vm1免密访问自己。
- 配置完成后，vm1可以自由访问自己和vm2，这对于集群管理和操作自动化非常重要。

4. 注意事项

避免多余的换行：在把公钥粘贴到authorized_keys文件时，不要加入多余的换行或空行，否则可能会导致连接问题。
不要覆盖原内容：如果authorized_keys里已经有其他密钥信息，不要把它们覆盖掉。否则可能会影响你对虚拟机的访问。

Part1：

任务解析

在这个任务中，你将使用Apache Hadoop分布式文件系统（HDFS）来创建一个分布式存储集群，并使用Apache Spark来处理数据。具体步骤包括安装Hadoop、配置HDFS和编辑相关配置文件。

1. HDFS和Spark的架构背景

Hadoop HDFS：主要用于存储大规模数据，通过数据分片并在多节点存储数据，以实现高可用性和容错性。
- NameNode：主节点，记录文件系统的元数据，比如文件位置和文件的基本信息。
- DataNode：存储数据块的从节点，每个DataNode都负责存储实际数据。
Spark：作为执行引擎，在分布式集群上并行处理数据。它将从HDFS中读取数据，进行快速计算和分析。

2. Part 1: 软件部署与配置步骤

Step 1：软件下载

在两台虚拟机（vm1 和 vm2）上下载Hadoop软件包。可以选择在vm1上下载后，用scp命令将文件传输到vm2上。

命令：

cd ~
wget https://archive.apache.org/dist/hadoop/common/hadoop-3.3.4/hadoop-3.3.4.tar.gz
tar zvxf hadoop-3.3.4.tar.gz
cd hadoop-3.3.4

解析：
- wget命令从Apache官网下载Hadoop的压缩包，tar zvxf用于解压文件。解压后，你将进入hadoop-3.3.4目录，方便后续配置。
core-site.xml
- 设置文件系统地址，使NameNode能在指定的IP地址（或主机名）上监听并管理文件系统。
Step 2：配置文件修改

在Hadoop中，有几个重要的配置文件需要设置，分别用于设置文件系统路径、数据节点目录、块大小和Java环境路径等。

Step 2: 核心配置文件
core-site.xml
- 设置文件系统地址，使NameNode能在指定的IP地址（或主机名）上监听并管理文件系统。示例：
```
<configuration><property><name>fs.default.name</name><value>hdfs://namenode_IP:9000</value></property>
</configuration>
```
- - 解释：这里的namenode_IP需要用主节点（即vm1的私有IPv4地址）替换，确保在集群内部通过私有网络传输数据，从而避免数据走公网。
hdfs-site.xml
- 设置NameNode和DataNode的本地存储目录，以及HDFS的默认块大小。
- 配置示例：
```
<configuration><property><name>dfs.namenode.name.dir</name><value>/path/to/namenode/dir/</value></property><property><name>dfs.datanode.data.dir</name><value>/path/to/datanode/dir</value></property><property><name>dfs.block.size</name><value>67108864</value><description>Block size</description></property>
</configuration>
```
  解释：dfs.namenode.name.dir用于指定NameNode存储路径，dfs.datanode.data.dir用于存储DataNode数据。块大小（dfs.block.size）设置为64MB，用于控制每个数据块的大小。此大小影响数据的分片和分布。
hadoop-env.sh
- 指定Hadoop的Java路径。
- 命令：
```
update-alternatives --display java
```
  将路径设置到hadoop-env.sh文件中的JAVA_HOME。
- 解释：运行上面的命令会显示Java安装路径，去掉/bin/java部分，然后将路径赋值给JAVA_HOME。这一步确保Hadoop能够找到Java环境，正常启动。
workers
- 在此文件中添加DataNode的IP地址，使NameNode能识别所有的DataNode节点。
- 修改方法：
  - 删除localhost行，添加vm1和vm2的私有IPv4地址。
- 解释：这一步告知Hadoop哪些节点会作为DataNode参与存储数据。删除localhost行，避免Hadoop尝试在本地查找DataNode。
- You also need to edit hadoop-3.3.4/etc/hadoop/workers to add the IP address of all the datanodes. Once again use the private IPv4 address. In our case for this assignment, you need to add two lines, each containing the IPv4 address of an EC2 instance corresponding to a datanode process. This way, the deployed HDFS can store data on both vm1 and vm2. (Delete the existing localhost line in this file.)

Step 3：将配置文件复制到vm2

操作命令：

scp hadoop-3.3.4/etc/hadoop/core-site.xml hadoop-3.3.4/etc/hadoop/hdfs-site.xml hadoop-3.3.4/etc/hadoop/workers hadoop-3.3.4/etc/hadoop/hadoop-env.sh ubuntu@<IP_of_vm2>:~/hadoop-3.3.4/etc/hadoop/

scp命令用于通过SSH连接将文件从本地机器传输到远程机器。这里将修改过的配置文件从vm1传输到vm2。
<IP_of_vm2>是vm2的私有IP地址（或公有IP地址）。ubuntu是远程服务器上的用户名。
解释：
配置文件在vm1上完成了所有更改，但在集群中所有节点（vm1和vm2）都需要这些配置，以便HDFS能识别所有节点并一致地存储数据。scp命令实现文件跨机器复制的作用，确保vm2获得相同配置。

Step 4：添加Hadoop路径并启动HDFS

在vm1上设置Hadoop二进制文件路径：
- 操作命令：
```
echo 'export PATH="/home/ubuntu/hadoop-3.3.4/bin:/home/ubuntu/hadoop-3.3.4/sbin:$PATH"' >> ~/.bashrc
source ~/.bashrc
```
  解释：
- echo命令将Hadoop的bin和sbin目录路径追加到.bashrc文件末尾。
- .bashrc是一个Shell配置文件，通常在用户登录时执行。将Hadoop路径添加到$PATH变量后，可以在终端直接调用Hadoop的命令而无需输入完整路径。
- source ~/.bashrc使更改立即生效，而无需重新登录。
格式化NameNode并启动HDFS守护进程：
- 命令：
```
hdfs namenode -format
start-dfs.sh
```
  解释：
- hdfs namenode -format：格式化NameNode的存储目录，清除任何旧数据，确保新配置生效。格式化操作只能在初次安装时执行，否则会删除现有的HDFS数据。
- start-dfs.sh：启动HDFS的所有守护进程（包括NameNode和DataNode）。NameNode运行在vm1上，而DataNode分别在vm1和vm2上运行。

检查HDFS的状态：

通过Web界面：
- 访问URL：<public_IPv4_DNS_address_of_vm1>:9870/dfshealth.html
- 解释：9870端口是HDFS的Web管理界面端口，可以在浏览器中查看HDFS状态，包括每个节点的运行状态、存储情况等。
通过命令行：
- 使用以下命令检查进程状态：
```
jps
```
  解释：jps命令列出Java进程。vm1上应显示NameNode和DataNode进程，而vm2上应显示DataNode进程，确保集群正确运行。

查看和管理HDFS命令：

命令：
```
hdfs dfs -help
```
解释：此命令展示了可用于管理HDFS的命令列表，例如创建目录、上传文件、查看文件等。

关闭HDFS集群：

命令：
```
stop-dfs.sh
```
解释：stop-dfs.sh命令用于停止集群中所有的HDFS进程。
复制配置文件：确保vm2节点上的配置文件与vm1一致。
设置路径：添加Hadoop路径使得命令行操作更便捷。
格式化并启动HDFS：格式化NameNode，启动整个分布式文件系统，确保数据可以在多个节点上进行存储和访问。
验证状态：通过Web界面和命令行检查进程，确认集群运行正常.

第三阶段：

任务解析

这一阶段将指导你在HDFS集群上部署和配置Apache Spark集群。以下是详细步骤和通俗解释，帮助你理解每一步的原理和操作目的。

Spark 概述

Apache Spark是一个开源的分析引擎，用于处理大规模数据。Spark的核心概念是Resilient Distributed Datasets (RDDs)，即可弹性分布式数据集。Spark集群有以下组件：

Master守护进程：主节点，负责资源分配，管理Worker节点。
Worker守护进程：工作节点，实际执行应用程序任务。
Executor：运行在Worker节点上的进程，负责具体的计算和数据存储。

在这个任务中，你将使用**独立模式（Standalone Mode）**部署Spark，这意味着Spark不需要依赖资源管理系统（如YARN）。

部署步骤

Step 1：下载Spark软件包

在集群的每个EC2实例上下载和解压Spark压缩包

wget https://archive.apache.org/dist/spark/spark-3.3.1/spark-3.3.1-bin-hadoop3.tgz
tar zvxf spark-3.3.1-bin-hadoop3.tgz

解析：
与Hadoop类似，这里我们先下载Spark的压缩包并解压到本地，以便进行后续配置。使用wget命令从官方存档下载Spark的安装包，解压后可以看到Spark的文件目录结构。

Step 2：修改配置文件

Spark的配置文件包含.template后缀，去除该后缀并进行相应的配置更改。

Python环境配置

在vm1和vm2的spark-env.sh文件中，添加以下内容：

export PYSPARK_PYTHON=/usr/bin/python3
export PYSPARK_DRIVER_PYTHON=/usr/bin/python3

解释：
PYSPARK_PYTHON和PYSPARK_DRIVER_PYTHON指定了Spark使用的Python解释器路径。由于Ubuntu 22.04自带Python 3.10.6，这一步确保Spark可以使用Python 3运行。

Workers配置

在vm1的spark-env.sh文件中，将私有IPv4地址添加到workers文件：

export SPARK_LOCAL_IP=<private_IPv4_address_of_vm1>
export SPARK_MASTER_HOST=<private_IPv4_address_of_vm1>

- 解释：
  SPARK_LOCAL_IP指定Spark在本节点绑定的IP地址，SPARK_MASTER_HOST指定主节点的IP地址，使其他节点可以识别并连接主节点。workers文件中添加的IP地址让Spark知道Worker节点的位置。

Worker节点配置

在workers文件中，添加vm1和vm2的私有IPv4地址：
```
<private_IPv4_address_of_vm1>
<private_IPv4_address_of_vm2>
```
解释：
这一步让Spark Master识别所有Worker节点，确保任务可以在多个节点上分布式执行。

启动Spark集群

在vm1上运行以下命令启动Spark集群：

spark-3.3.1-bin-hadoop3/sbin/start-all.sh

Web监控：在浏览器中访问<public_IPv4_DNS_address_of_vm1>:8080以查看Spark集群状态。
进程检查：使用jps命令检查vm1是否有Master进程在运行，而vm1和vm2上是否都有Worker进程在运行。

解释：
start-all.sh启动Spark的所有守护进程（Master和Worker），并在集群中分布执行任务。在浏览器中访问8080端口的Spark管理页面，可以查看Spark的资源状态和各个节点的工作情况。

资源配置

在spark-defaults.conf中配置Spark应用程序的内存和CPU资源：

配置示例：

spark.master                   spark://<private_IPv4_address_of_vm1>:7077
spark.driver.memory            4g
spark.executor.memory          4g
spark.executor.cores           2
spark.task.cpus                1

解释：

spark.master：指定Spark Master的IP地址和端口（7077端口用于Spark集群通信）。
spark.driver.memory和spark.executor.memory：分别为Driver和Executor分配4GB内存。由于t3.large实例只有8GB内存，这个设置可以使集群性能最大化。
spark.executor.cores：每个Executor使用2个CPU核，适应实例的2 vCPU。
spark.task.cpus：每个任务使用1个CPU核。

停止Spark集群

使用以下命令停止Spark集群：

spark-3.3.1-bin-hadoop3/sbin/stop-all.sh

解释：
stop-all.sh命令会停止集群中所有的Spark进程，使集群处于待机状态。

Part 2

在此任务中，你将编写一个简单的Spark应用程序实现词频统计（Word Count），并将结果存储到HDFS中。以下是步骤说明、操作指南以及通俗解释，帮助你理解并成功完成这个应用程序的开发和部署。

任务概述：
下载样例数据到vm1的主目录。
将数据加载到HDFS。
使用PySpark编写一个Python程序来统计每个单词的出现次数。
将输出结果保存为CSV格式文件存储在HDFS中。
部署和开发步骤：

Step 1：下载数据文件

将提供的链接中的数据文件下载到vm1的主目录下，文件名为wordcount.txt。

Assignment-4/wordcount.txt at main · CUHKSZ-CSC4160-Fall24/Assignment-4https://github.com/CUHKSZ-CSC4160-Fall24/Assignment-4/blob/main/wordcount.txt
Step 2：将文件加载到HDFS

在vm1中运行以下命令，将文件上传到HDFS根目录（/）中：

hdfs dfs -copyFromLocal wordcount.txt /

解析： hdfs dfs -copyFromLocal命令用于将本地文件传输到HDFS上。上传文件后，可以通过Spark从HDFS中读取文件，进行分布式处理。

Step 3：编写Word Count程序 PySpark编程框架

你需要创建一个SparkSession，这是进入Spark所有功能的入口。

示例代码：

from pyspark.sql import SparkSession# 创建SparkSession
spark = (SparkSession.builder.appName("Word Count Program").getOrCreate())# 读取数据文件
df = spark.read.text("hdfs://<IP_of_vm1>:9000/wordcount.txt")# 处理数据，分割单词
words = df.selectExpr("explode(split(value, ' ')) as word")# 计算每个单词的频次
word_counts = words.groupBy("word").count()# 将结果保存为CSV格式到HDFS
word_counts.write.csv("hdfs://<IP_of_vm1>:9000/output")# 结束SparkSession
spark.stop()

代码解析：

spark.read.text(...)：从HDFS中读取文件。 split(value, ' ')：基于空格将文本分割为单词。 explode(...)：将分割后的单词展开为行。 groupBy(...).count()：按单词分组并计算出现次数。 write.csv(...)：将统计结果以CSV格式保存到HDFS中。代码结构：

将以上代码另存为your_pyspark_app_part2.py，并将其保存在/path/to/your_pyspark_app_part2.py路径下。

Step 4：提交Spark应用程序使用spark-submit命令将Python程序提交给Spark运行：

bin/spark-submit /path/to/your_pyspark_app_part2.py hdfs://<IP_of_vm1>:9000/wordcount.txt hdfs://<IP_of_vm1>:9000/output

解析： spark-submit命令是Spark应用程序提交的入口。通过它，Spark将会启动应用程序，将wordcount.txt文件作为输入，并将输出结果存储到output目录下。

Step 5：查看应用程序状态访问Spark UI：在浏览器中访问

<public_IPv4_DNS_address_of_vm1>:8080，可以查看Spark集群的整体运行情况，包括应用程序的ID。

查看应用程序详情：

点击Spark UI中的应用程序ID，进入应用程序状态页面，查看DAG可视化、执行统计等详细信息。

注意：如果在EC2中遇到“无法连接服务器”问题，请将URL中的私有IPv4地址替换为vm1的公有IPv4 DNS地址，并保留端口号:4040。

总结

数据上传到HDFS：将文本文件上传到HDFS，使其可以被Spark分布式处理。

编写并运行Word Count应用：使用Spark SQL和DataFrame API实现对文本文件中单词的统计。

结果输出到HDFS：将结果存储为CSV文件在HDFS中，便于进一步查看和处理。

监控应用状态：通过Spark UI监控应用的执行情况并获取详细的执行信息。

这样，通过上述步骤，你的Spark应用程序将实现一个完整的词频统计功能，并在分布式环境中高效运行。

Part 3: '

任务解析

在这个任务中，你将使用PySpark实现PageRank算法，并在分布式集群环境中进行性能优化和容错测试。以下是详细步骤、操作指南以及相关解释，帮助你理解每个任务的核心内容。

PageRank 算法概述

PageRank是一种基于链接分析的算法，最初由Google用于评价网页的重要性。它的基本思路是：一个页面的重要性与指向它的链接数量和质量相关。具体来说：

初始分数：每个页面的初始排名分数设置为1。
贡献计算：对于每个页面ppp，它会将其排名分数均匀地分配给其所有的出链邻居。贡献值 = rank(p) / # 出链数
排名更新：每个页面的新的排名分数 = 0.15 + 0.85 * sum(贡献值)
迭代：重复上述步骤，直到算法收敛或达到指定迭代次数（这里设为10次）。

在此任务中，你将使用Berkeley-Stanford网页图数据集，每行表示一个页面及其邻居。

In this part, you will implement the PageRank algorithm (§2.1) (or the Wikipedia version), which is an algorithm used by search engines like Google to evaluate the quality of links to a webpage. The algorithm can be summarized as follows:

Set initial rank of each page to be 1.
On each iteration, each page p contributes to its outgoing neighbors a value of rank(p)/(# of outgoing neighbors of p).
Update each page’s rank to be 0.15 + 0.85 * (sum of contributions).
Go to next iteration.

In this assignment, we will run the algorithm on the Berkeley-Stanford web graph dataset. We have also upload the decompressed dataset on the BlackBoard in case the original website is unavailable. Each line in the dataset consists of a page and one of its neighbors. You need to decompress the downloaded .gz dataset (using gunzip) and copy the input file to HDFS first. In this part, always run the algorithm for a total of 10 iterations.

Task 1. Write a PySpark application that implements the PageRank algorithm.
Task 2. In order to achieve high parallelism, Spark will split the data into smaller chunks called partitions, which are distributed across different nodes in the cluster. Partitions can be changed in several ways. For example, any shuffle operation on an RDD (e.g., join()) will result in a change in partitions (customizable via user’s configuration). In addition, one can also decide how to partition data when creating/configuring RDDs (hint: e.g., you can use the function partitionBy()). For this task, add appropriate custom RDD partitioning and see what changes.
Task 3. Kill a Worker process and see the changes. You should trigger the failure to a selected worker VM when the application reaches anywhere between 25% to 75% of its lifetime (hint: use the Application Detail UI to track the application progress):

Clear the memory cache using sudo sh -c "sync; echo 3 > /proc/sys/vm/drop_caches" on vm2;
Kill the Worker process on vm2: Use jps to get the process ID (PID) of the Spark Worker on vm2 and then use the command kill -9 <Worker_PID> to kill the Spark Worker process.

常见问题解答（FAQs）

如果您在访问HDFS/Spark的Web仪表板时遇到问题，请尝试以下故障排除方法：

您应该使用公网IPv4 DNS地址，后面加上冒号和端口号，而不是使用公网IPv4地址或私网地址。如果这样做不起作用，请检查您是否正确配置了spark-3.3.1-bin-hadoop3/conf/spark-env.sh文件。如果仍然不起作用，请检查您在创建EC2实例时设置的安全组。作为最后的手段，您可以创建一个新的安全组，允许所有流量，设置为0.0.0.0/0，然后将其附加到vm1和vm2实例（仅配置vm1将导致Spark作业挂起，HDFS无法复制），并从vm1和vm2中移除旧的安全组。

如果您需要/想要使用Jupyter Notebook来完成此作业，请参考笔记本设置教程。

如果您想使用GitHub来保存您的进度，请参考这篇文章 this post。