AWS EC2 搭建 Hadoop 和 Spark 集群

前言

本篇演示如何使用 AWS EC2 雲服務搭建集群。當然在只有一台計算機的情況下搭建完全分布式集群，還有另外幾種方法：一種是本地搭建多台虛擬機，好處是免費易操控，壞處是虛擬機對宿主機配置要求較高； 另一種方案是使用 AWS EMR ，是亞馬遜專門設計的集群平台，能快速啟動集群，且具有較高的靈活性和擴展性，能方便地增加機器。然而其缺點是只能使用預設的軟件，如下圖：

如果要另外裝軟件，則需要使用 Bootstrap 腳本，詳見 https://docs.aws.amazon.com/zh_cn/emr/latest/ManagementGuide/emr-plan-software.html?shortFooter=true ，可這並不是一件容易的事情，很多次軟件都裝不上去。 另外，如果在 EMR 上關閉了集群，則里面的文件和配置都不會保存，下次使用時全部要重新設置，可見其比較適用於一次性使用的場景。

綜上所述，如果使用純 EC2 進行手工搭建，則既不會受本地資源限制，也具有較高的靈活性，可以隨意配置安裝軟件。而其缺點就是要手工搭建要耗費較多時間，而且在雲上操作和在本地操作有些地方是不一樣的，只要有一步出錯可能就要卡殼很久，鑒於網上用 EC2 搭建這方面資料很少，因此這里寫一篇文章把主要流程記錄下來。

如果之前沒有使用過 EC2，可能需要花一段時間熟悉，比如注冊以及創建密鑰對等步驟，官方提供了相關教程 。另外我的本地機和雲端機使用的都是 Ubuntu 16.04 LTS 64位，如果你的本地機是 Windows，則需要用 Git 或 PuTTY 連接雲端機，詳情參閱 https://docs.aws.amazon.com/zh_cn/AWSEC2/latest/UserGuide/putty.html 。

創建 EC2 實例

下面正式開始，這里設立三台機器 (實例)，一台作主節點 (master node)，兩台作從節點 (slaves node)。首先創建實例，選擇 Ubuntu Server 16.04 LTS (HVM) ，實例類型選擇價格低廉的 t2.medium 。如果是第一次用，就不要選價格太高的類型了，不然萬一操作失誤了每月賬單可承受不起。

在第 3 步中，因為要同時開三台機器，Number of Instances 可以直接選擇3。但如果是每台分別開的話，下面的 Subnet 都要選擇同一個區域，不然機器間無法通信，詳情參閱 https://docs.aws.amazon.com/zh_cn/AWSEC2/latest/UserGuide/using-regions-availability-zones.html 。

第 4 步設置硬盤大小，如果就搭個集群可能不用動，如果還要裝其他軟件，可能就需要在這里增加容量了，我是增加到了 15 GB：

第 5 和第 6 步直接Next 即可，到第 7 步 Launch 后選擇或新建密鑰對，就能得到創建好的 3 個實例，這里可以設置名稱備注，如 master、slave01、slave02 等：

開啟 3 個終端窗口，ssh 連接3個實例，如 ssh -i xxxx.pem [e-mail] ，其中 xxxx.pem 是你的本地密鑰對名稱，ec2-xx-xxx-xxx-xx.us-west-2.compute.amazonaws.com 是該實例的外部 DNS 主機名，每台實例都不一樣。這里需要說明一下，因為這是和本地開虛擬機的不同之處： EC2 的實例都有公有 IP和私有 IP之分，私有 IP 用於雲上實例之間的通信，而公有 IP 則用於你的本地機與實例之間的通信，因此這里 ssh 連接使用的是公有 IP (DNS) 。在下面搭建集群的步驟中也有需要填寫公有和私有 IP ，注意不要填反了。關於二者的區別參閱 https://docs.aws.amazon.com/zh_cn/AWSEC2/latest/UserGuide/using-instance-addressing.html?shortFooter=true#using-instance-addressing-common 。

新增 hadoop 用戶、安裝 Java 環境

以下以 master 節點為例。登陸實例后，默認用戶為 ubuntu，首先需要創建一個 hadoop 用戶：

$ sudo useradd -m hadoop -s /bin/bash   # 增加 hadoop用戶
$ sudo passwd hadoop  			        # 設置密碼，需要輸入兩次
$ sudo adduser hadoop sudo  		    # 為 hadoop 用戶增加管理員權限
$ su hadoop   				            # 切換到 hadoop 用戶，需要輸入密碼
$ sudo apt-get update  			        # 更新 apt 源

這一步完成之后，終端用戶名會變為 hadoop，且 /home 目錄下會另外生成一個 hadoop 文件夾。

Hadoop 依賴於 Java 環境，所以接下來需要先安裝 JDK，直接從官網下載，這里下的是 Linux x64 版本 jdk-8u231-linux-x64.tar.gz ，用 scp 遠程傳輸到 master 機。注意這里只能傳輸到 ubuntu 用戶下，傳到 hadoop 用戶下可能會提示權限不足。

$ scp -i xxx.pem jdk-8u231-linux-x64.tar.gz [e-mail]:/home/ubuntu/  # 本地執行該命令

本篇假設所有軟件都安裝在 /usr/lib 目錄下：

$ sudo mv /home/ubuntu/jdk-8u231-linux-x64.tar.gz /home/hadoop         # 將文件移動到 hadoop 用戶下
$ sudo tar -zxf /home/hadoop/jdk-8u231-linux-x64.tar.gz -C /usr/lib/   # 把JDK文件解壓到/usr/lib目錄下
$ sudo mv /usr/lib/jdk1.8.0_231  /usr/lib/java                         # 重命名java文件夾
$ vim ~/.bashrc 								                       # 配置環境變量，貌似EC2只能使用 vim

添加如下內容：

export JAVA_HOME=/usr/lib/java
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

$ source ~/.bashrc   # 讓配置文件生效
$ java -version    # 查看 Java 是否安裝成功

如果出現以下提示則表示安裝成功：

在 master 節點完成上述步驟后，在兩個 slave 節點完成同樣的步驟 (新增 hadoop 用戶、安裝 Java 環境)

網絡配置

這一步是為了便於 Master 和 Slave 節點進行網絡通信，在配置前請先確定是以 hadoop 用戶登錄的。首先修改各個節點的主機名，執行 sudo vim /etc/hostname ，在 master 節點上將 ip-xxx-xx-xx-xx 變更為 Master 。其他節點類似，在 slave01 節點上變更為 Slave01，slave02 節點上為 Slave02。

然后執行 sudo vim /etc/hosts 修改自己所用節點的IP映射，以 master 節點為例，添加紅色區域內信息，注意這里的 IP 地址是上文所述的私有 IP：

接着在兩個 slave 節點的hosts中添加同樣的信息。完成后重啟一下，在進入 hadoop 用戶，能看到機器名的變化 (變成 Master 了)：

對於 ec2 實例來說，還需要配置安全組 (Security groups)，使實例能夠互相訪問 ：

選擇划線區域，我因為是同時建立了三台實例，所以安全組都一樣，如果不是同時建立的，這可能三台都要配置。

進入后點擊 Inbound 再點 Edit ，再點擊 Add Rule，選擇里面的 All Traffic ，接着保存退出：

三台實例都設置完成后，需要互相 ping 一下測試。如果 ping 不通，后面是不會成功的：

$ ping Master -c 3   # 分別在3台機器上執行這三個命令
$ ping Slave01 -c 3
$ ping Slave02 -c 3

接下來安裝 SSH server， SSH 是一種網絡協議，用於計算機之間的加密登錄。安裝完 SSH 后，要讓 Master 節點可以無密碼 SSH 登陸到各個 Slave 節點上，在Master節點執行：

$ sudo apt-get install openssh-server
$ ssh localhost                                         # 使用 ssh 登陸本機，需要輸入 yes 和 密碼
$ exit                                                  # 退出剛才的 ssh localhost, 注意不要退出hadoop用戶
$ cd ~/.ssh/                                            # 若沒有該目錄，請先執行一次ssh localhost
$ ssh-keygen -t rsa                                     # 利用 ssh-keygen 生成密鑰，會有提示，瘋狂按回車就行
$ cat ./id_rsa.pub >> ./authorized_keys                 # 將密鑰加入授權
$ scp ~/.ssh/id_rsa.pub Slave01:/home/hadoop/           # 將密鑰傳到 Slave01 節點
$ scp ~/.ssh/id_rsa.pub Slave02:/home/hadoop/           # 將密鑰傳到 Slave02 節點

接着在 Slave01和 Slave02 節點上，將 ssh 公匙加入授權：

$ mkdir ~/.ssh       # 如果不存在該文件夾需先創建，若已存在則忽略
$ cat ~/id_rsa.pub >> ~/.ssh/authorized_keys

這樣，在 Master 節點上就可以無密碼 SSH 到各個 Slave 節點了，可在 Master 節點上執行如下命令進行檢驗，如下圖所示變為 Slave01了，再按 exit 可退回到 Master：

至此網絡配置完成。

安裝 Hadoop

去到鏡像站 https://archive.apache.org/dist/hadoop/core/ 下載，我下載的是 hadoop-2.8.4.tar.gz 。在 Master 節點上執行：

$ sudo tar -zxf /home/ubuntu/hadoop-2.8.4.tar.gz -C /usr/lib     # 解壓到/usr/lib中
$ cd /usr/lib/
$ sudo mv ./hadoop-2.8.4/ ./hadoop                               # 將文件夾名改為hadoop
$ sudo chown -R hadoop ./hadoop                                  # 修改文件權限

將 hadoop 目錄加到環境變量，這樣就可以在任意目錄中直接使用 hadoop、hdfs 等命令。執行 vim ~/.bashrc ，加入一行：

export PATH=$PATH:/usr/lib/hadoop/bin:/usr/lib/hadoop/sbin

保存后執行 source ~/.bashrc 使配置生效。

完成后開始修改 Hadoop 配置文件(這里也順便配置了 Yarn)，先執行 cd /usr/lib/hadoop/etc/hadoop ，共有 6 個需要修改 —— hadoop-env.sh、slaves、core-site.xml、hdfs-site.xml、mapred-site.xml、yarn-site.xml 。

1、文件 hadoop-env.sh 中把 export JAVA_HOME=${JAVA_HOME} 修改為 export JAVA_HOME=/usr/lib/java ，即 Java 安裝路徑。

2、 文件 slaves 把里面的 localhost 改為 Slave01和 Slave02 。

3、core-site.xml 改為如下配置：

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

4、hdfs-site.xml 改為如下配置：

<configuration>
        <property>
                <name>dfs.namenode.secondary.http-address</name>
                <value>Master:50090</value>
        </property>
        <property>
                <name>dfs.replication</name>
                <value>2</value>
        </property>
        <property>
                <name>dfs.namenode.name.dir</name>
                <value>file:/usr/lib/hadoop/tmp/dfs/name</value>
        </property>
        <property>
                <name>dfs.datanode.data.dir</name>
                <value>file:/usr/lib/hadoop/tmp/dfs/data</value>
        </property>
</configuration>

5、文件 mapred-site.xml (可能需要先重命名，默認文件名為 mapred-site.xml.template)：

<configuration>
        <property>
                <name>mapreduce.framework.name</name>
                <value>yarn</value>
        </property>
        <property>
                <name>mapreduce.jobhistory.address</name>
                <value>Master:10020</value>
        </property>
        <property>
                <name>mapreduce.jobhistory.webapp.address</name>
                <value>Master:19888</value>
        </property>
</configuration>

6、文件 yarn-site.xml ：

<configuration>
        <property>
                <name>yarn.resourcemanager.hostname</name>
                <value>Master</value>
        </property>
        <property>
                <name>yarn.nodemanager.aux-services</name>
                <value>mapreduce_shuffle</value>
        </property>
</configuration>

配置好后，將 Master 上的 /usr/lib/hadoop 文件夾復制到各個 slave 節點上。在 Master 節點上執行：

$ cd /usr/lib
$ tar -zcf ~/hadoop.master.tar.gz ./hadoop   # 先壓縮再復制
$ scp ~/hadoop.master.tar.gz Slave01:/home/hadoop
$ scp ~/hadoop.master.tar.gz Slave02:/home/hadoop

分別在兩個 slave 節點上執行：

$ sudo tar -zxf ~/hadoop.master.tar.gz -C /usr/lib
$ sudo chown -R hadoop /usr/lib/hadoop

安裝完成后，首次啟動需要先在 Master 節點執行 NameNode 的格式化：

$ hdfs namenode -format       # 首次運行需要執行初始化，之后不需要

成功的話，會看到 “successfully formatted” 和 “Exitting with status 0” 的提示，若為 “Exitting with status 1” 則是出錯。

接着可以啟動 Hadoop 和 Yarn 了，啟動需要在 Master 節點上進行：

$ start-dfs.sh
$ start-yarn.sh
$ mr-jobhistory-daemon.sh start historyserver

通過命令 jps 可以查看各個節點所啟動的進程。正確的話，在 Master 節點上可以看到 NameNode、ResourceManager、SecondrryNameNode、JobHistoryServer 進程，如下圖所示：

在 Slave 節點可以看到 DataNode 和 NodeManager 進程，如下圖所示：

通過命令 hdfs dfsadmin -report 可查看集群狀態，其中 Live datanodes (2) 表明兩個從節點都已正常啟動，如果是 0 則表示不成功：

可以通過下列三個地址查看 hadoop 的 web UI，其中 ec2-xx-xxx-xxx-xx.us-west-2.compute.amazonaws.com 是該實例的外部 DNS 主機名，50070、8088、19888 分別是 hadoop、yarn、JobHistoryServer 的默認端口：

ec2-xx-xxx-xxx-xx.us-west-2.compute.amazonaws.com:50070
ec2-xx-xxx-xxx-xx.us-west-2.compute.amazonaws.com:8088
ec2-xx-xxx-xxx-xx.us-west-2.compute.amazonaws.com:19888

執行 Hadoop 分布式實例

$ hadoop fs -mkdir -p /user/hadoop   # 在hdfs上創建hadoop賬戶
$ hadoop fs -mkdir input
$ hadoop fs -put /usr/lib/hadoop/etc/hadoop/*.xml input  # 將hadoop配置文件復制到hdfs中
$ hadoop jar /usr/lib/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-*.jar grep input output 'dfs[a-z.]+'  # 運行實例

如果成功可以看到以下輸出：

最后關閉 Hadoop 集群需要執行以下命令：

$ stop-yarn.sh
$ stop-dfs.sh
$ mr-jobhistory-daemon.sh stop historyserver

安裝 Spark

去到鏡像站 https://archive.apache.org/dist/spark/ 下載，由於之前已經安裝了Hadoop，所以我下載的是無 Hadoop 版本的，即 spark-2.3.3-bin-without-hadoop.tgz 。在 Master 節點上執行：

$ sudo tar -zxf /home/ubuntu/spark-2.3.3-bin-without-hadoop.tgz -C /usr/lib  # 解壓到/usr/lib中
$ cd /usr/lib/
$ sudo mv ./spark-2.3.3-bin-without-hadoop/ ./spark  # 將文件夾名改為spark
$ sudo chown -R hadoop ./spark						  # 修改文件權限

將 spark 目錄加到環境變量，執行 vim ~/.bashrc 添加如下配置：

export SPARK_HOME=/usr/lib/spark
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin

保存后執行 source ~/.bashrc 使配置生效。

接着需要配置了兩個文件，先執行 cd /usr/lib/spark/conf 。

1、 配置 slaves 文件

mv slaves.template slaves  # 將slaves.template重命名為slaves

slaves文件設置從節點。編輯 slaves 內容,把默認內容localhost替換成兩個從節點的名字：

Slave01
Slave02

2、配置 spark-env.sh 文件

mv spark-env.sh.template spark-env.sh

編輯 spark-env.sh 添加如下內容：

export SPARK_DIST_CLASSPATH=$(/usr/lib/hadoop/bin/hadoop classpath)
export HADOOP_CONF_DIR=/usr/lib/hadoop/etc/hadoop
export SPARK_MASTER_IP=172.31.40.68   # 注意這里填的是Master節點的私有IP 
export JAVA_HOME=/usr/lib/java

配置好后，將 Master 上的 /usr/lib/spark 文件夾復制到各個 slave 節點上。在 Master 節點上執行：

$ cd /usr/lib
$ tar -zcf ~/spark.master.tar.gz ./spark
$ scp ~/spark.master.tar.gz Slave01:/home/hadoop
$ scp ~/spark.master.tar.gz Slave02:/home/hadoop

然后分別在兩個 slave 節點上執行：

$ sudo tar -zxf ~/spark.master.tar.gz -C /usr/lib
$ sudo chown -R hadoop /usr/lib/spark

在啟動 Spark 集群之前，先確保啟動了 Hadoop 集群：

$ start-dfs.sh
$ start-yarn.sh
$ mr-jobhistory-daemon.sh start historyserver
$ start-master.sh  # 啟動 spark 主節點
$ start-slaves.sh  # 啟動 spark 從節點

可通過 ec2-xx-xxx-xxx-xx.us-west-2.compute.amazonaws.com:8080 訪問 spark web UI 。

執行 Spark 分布式實例

1、通過命令行提交 JAR 包：

$ spark-submit --class org.apache.spark.examples.SparkPi --master spark://Master:7077 /usr/lib/spark/examples/jars/spark-examples_2.11-2.3.3.jar 100 2>&1 | grep "Pi is roughly" 

結果如下說明成功：

2、通過 IDEA 遠程連接運行程序：

可以在 本地 IDEA 中編寫代碼，遠程提交到雲端機上執行，這樣比較方便調試。需要注意的是 Master 地址填雲端機的公有 IP 地址。下面以一個 WordVec 程序示例，將句子轉換為向量形式：

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.log4j.{Level, Logger}
import org.apache.spark.ml.feature.Word2Vec
import org.apache.spark.ml.linalg.Vector
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession

object Word2Vec {
  def main(args: Array[String]) {
    Logger.getLogger("org").setLevel(Level.ERROR)  // 控制輸出信息
    Logger.getLogger("com").setLevel(Level.ERROR)

    val conf = new SparkConf()
      .setMaster("spark://ec2-54-190-51-132.us-west-2.compute.amazonaws.com:7077")  // 填公有DNS或公有IP地址都可以
      .setAppName("Word2Vec")
      .set("spark.cores.max", "4")
      .set("spark.executor.memory", "2g")
    val sc = new SparkContext(conf)

    val spark = SparkSession
      .builder
      .appName("Word2Vec")
      .getOrCreate()

    val documentDF = spark.createDataFrame(Seq(
      "Hi I heard about Spark".split(" "),
      "I wish Java could use case classes".split(" "),
      "Logistic regression models are neat".split(" ")
    ).map(Tuple1.apply)).toDF("text")

    val word2Vec = new Word2Vec()
      .setInputCol("text")
      .setOutputCol("result")
      .setVectorSize(3)
      .setMinCount(0)
    val model = word2Vec.fit(documentDF)

    val result = model.transform(documentDF)
    result.collect().foreach { case Row(text: Seq[_], features: Vector) =>
      println(s"Text: [${text.mkString(", ")}] => \nVector: $features\n") }
  }
}

IDEA 控制台輸出：

關閉 Spark 和 Hadoop 集群有以下命令：

$ stop-master.sh
$ stop-slaves.sh
$ stop-yarn.sh
$ stop-dfs.sh
$ mr-jobhistory-daemon.sh stop historyserver

/