環境變數配置

通過命令列引數配置

通過spark-env.sh配置

也可以通過在conf / spark-env.sh中設定環境變數來進一步配置叢集，不過在該檔案中配置的屬性優先順序沒有命令列引數高。

1. master環境變數
   
   • SPARK_MASTER_HOST：將Master繫結到特定的主機名或IP地址；
   • SPARK_MASTER_PORT：在指定的埠啟動Master（預設：7077）；
   • SPARK_MASTER_WEBUI_PORT：Master Web UI的埠（預設：8080）；
   • SPARK_MASTER_OPTS：以“-Dx = y”（預設值：無）格式應用於Master的配置屬性，具體支援的屬性如下configuration.html#SPARK_MASTER_OPTS
2. worker環境變數
   
   • SPARK_WORKER_CORES：允許Spark Applications在機器上使用的核心總數（預設：所有可用核心）；
   • SPARK_WORKER_MEMORY：允許Spark Applications在機器上使用的記憶體總數（預設：所有1G）。請注意，每個Application使用的記憶體都使用其spark.executor.memory屬性進行配置。
   • SPARK_WORKER_PORT：Worker的啟用特定埠（預設值：隨機分配）；
   • SPARK_WORKER_WEBUI_PORT：Worker的WebUI埠號（預設值是8081）；
   • SPARK_WORKER_DIR：在Worker中執行Application的目錄，其中將包括日誌和暫存空間（預設值：SPARK_HOME / work）。
   • SPARK_WORKER_OPTS：以“-Dx = y”（預設值：無）格式應用於Worker的配置屬性，具體支援的屬性如下configuration.html#SPARK_WORKER_OPTS
3. 通用的環境變數
   
   • SPARK_LOCAL_DIRS：用於儲存map輸出檔案和儲存在磁碟上的RDD快取。這應該在系統中的快速本地磁碟上，它也可以是不同磁碟上多個目錄的逗號分隔列表。
   • SPARK_DAEMON_MEMORY：分配給Master和Workers守護程序的記憶體（預設是1G）；
   • SPARK_DAEMON_JAVA_OPTS：Spark master和worker守護程序的JVM選項，格式為“-Dx = y”（預設值：無）。
   • SPARK_PUBLIC_DNS：Spark Worker和Master的公共DNS名稱。

叢集的啟動過程

Master和Workers需要通過指令碼來啟動，可以使用start-all.sh一鍵啟動Master和Workers，也可以使用start-master.sh和start-slaves.sh分別啟動Master和Workers。總之，不管用什麼指令碼啟動，它們之間的呼叫鏈如下圖所示：

Master的啟動

指令碼的執行流程

1. start-master.sh

   使用start-master.sh，在指令碼執行的機器上啟動Master例項，start-master.sh指令碼的主要邏輯包括：

   • 確定了master的啟動類org.apache.spark.master.Master；
   • 通過執行spark-env.sh，獲取到了master的執行環境變數，包括使用者配置的主機地址、埠號和WebUI的埠號。如果沒有配置，則使用預設配置。
   • 最後，將以上引數提交給 spark-daemon.sh，啟動一個Master守護程序，具體的執行命令如下：

   "${SPARK_HOME}/sbin"/spark-daemon.sh start $CLASS 1 \
    --host $SPARK_MASTER_HOST --port $SPARK_MASTER_PORT --webui-port $SPARK_MASTER_WEBUI_PORT \
    $ORIGINAL_ARGS

2. start-daemon.sh

   spark-daemon.sh為守護程序提供了一些管理操作，包括啟動、停止、檢視執行狀態等，具體用法如下：

   
   # start|stop|submit|status : 表示option
   
   
   # spark-command : 表示啟動命令，如org.apache.spark.master.Master
   
   
   # spark-instance-number：啟動的例項數，一般為1，對於Slave可以在一個節點上啟動多個Worker；
   
   
   # args...：表示提供給spark-command 的執行引數。
   
   $ spark-daemon.sh [--config <conf-dir>] (start|stop|submit|status) <spark-command> <spark-instance-number> <args...>

   spark-daemon.sh指令碼執行的流程主要包括以下幾個步驟：

   • 解析命令列引數，得到option、command和instance；

   • 初始化一些環境變數，包括日誌目錄、pid檔案儲存目錄、排程的優先順序、是否為daemon啟動一個單獨的程序，還是依附於執行指令碼的bash所在的程序等；

   • 根據不同的option執行對應的操作，如果是stop或status，則根據pid檔案執行對應的操作。如果是submit和start，接著會呼叫run_command方法：

     case $option in
     
     (submit)
      run_command submit "[email protected]"
      ;;
     
     (start)
      run_command class "[email protected]"
      ;;

     run_command() {
     mode="$1"
     shift
     
     mkdir -p "$SPARK_PID_DIR"
     
     
     # 根據pid檔案檢查是否已經正在執行
     
     if [ -f "$pid" ]; then
      TARGET_ID="$(cat "$pid")"
      if [[ $(ps -p "$TARGET_ID" -o comm=) =~ "java" ]]; then
        echo "$command running as process $TARGET_ID.  Stop it first."
        exit 1
      fi
     fi
     
     if [ "$SPARK_MASTER" != "" ]; then
      echo rsync from "$SPARK_MASTER"
      rsync -a -e ssh --delete --exclude=.svn --exclude='logs/*' --exclude='contrib/hod/logs/*' "$SPARK_MASTER/" "${SPARK_HOME}"
     fi
     
     # 日誌遷移，將舊的日誌編號遞增，保證每次啟動後日志名稱都為 **.Master-1-master0.out，而舊的日誌名稱則為 .Master-1-master0.out.1 .Master-1-master0.out.2
     
     spark_rotate_log "$log"
     echo "starting $command, logging to $log"
     
     
     # 根據不同的模式，確定執行命令所使用的指令碼
     
     case "$mode" in
      (class)
        execute_command nice -n "$SPARK_NICENESS" "${SPARK_HOME}"/bin/spark-class "$command" "[email protected]"
        ;;
     
      (submit)
        execute_command nice -n "$SPARK_NICENESS" bash "${SPARK_HOME}"/bin/spark-submit --class "$command" "[email protected]"
        ;;
     
      (*)
        echo "unknown mode: $mode"
        exit 1
        ;;
     esac
     
     }

     由於本次的執行模式mode是class，最終將執行命令交給了spark-class，來啟動的可執行類。

3. spark-class

   spark-class指令碼會載入spark配置的環境變數資訊spark-env.sh、獲取類路徑LAUNCH_CLASSPATH（預設是${SPARK_HOME}/jars）、選擇執行此命令的執行器（這裡是java），並呼叫build_command構建主類的啟動命令。

   build_command() {
    "$RUNNER" -Xmx128m -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "[email protected]"
    printf "%d\0" $? # 代表上一個命令執行是否成功的標誌，如果執行成功則為0，否則不為0
   }

   可以看出，構建主類啟動命令是通過org.apache.spark.launcher.Main類進行的，Main類作用是接收[class] [class args]格式的引數，並且提供了一下兩種工作模式：

   • 如果使用的是spark-submit指令碼提交，那麼class一定為org.apache.spark.deploy.SparkSubmit，生成的啟動命令是：

     /opt/jdk/1.8.0_151/bin/java -cp /opt/spark/2.2.0/conf/:/opt/spark/2.2.0/jars/* -Xmx 1g org.apache.spark.deploy.SparkSubmit --master spark://master0:7077 --deploy-mode cluster --class cn.vibrancy.spark.ComplexJob --name complexjob /home/deploy/complex-job.jar

   • 如果執行的是spark-class指令碼，那麼執行引數提供的class，生成的啟動命令是：

     /opt/jdk/1.8.0_151/bin/java -cp /opt/spark/2.2.0/conf/:/opt/spark/2.2.0/jars/* -Xmx 1g org.apache.spark.master.Master -host master -p 7077 --webui-port 8080

   接著，將構建出的主class啟動命令列印，bash可以使用read來獲取列印的結果，並放到CMD中。

   // Main.java
   // 將生成的命令打印出，bash可以獲取並收集到列印的結果
   List<String> bashCmd = prepareBashCommand(cmd, env);
   for (String c : bashCmd) {
    System.out.print(c);
    System.out.print('\0');
   }

   
   # spark-class
   
   set +o posix
   CMD=()
   while IFS= read -d '' -r ARG; do
    CMD+=("$ARG")
   done < <(build_command "[email protected]")

   最後，執行CMD命令，這樣就可以執行Master的main方法了。

   
   # spark-class
   
   CMD=("${CMD[@]:0:$LAST}")
   exec "${CMD[@]}"

Master的初始化

1. 建立RpcEnv，並向RpcEnv例項中註冊當前MasterEndpoint；
2. 初始化Worker、Application和Driver相關的資料結構；
3. 根據RECOVERY_MODE建立對應的Master元資料持久化引擎和領導選舉機制，包括ZOOKEEPER、FILESYSTEM、CUSTOMER和NONE；
4. 等待Worker的註冊；

Worker的啟動

指令碼的執行流程

可以在執行start-slave.sh指令碼所在機器上啟動一個Worker，也可以在任意一個節點上執行start-slaves.sh指令碼，則會啟動slaves中記錄的所有從節點上的Worker。

1. start-slaves.sh

   在任意節點上啟動叢集上所有的Worker原理很簡單，在start-slaves.sh中：

   
   # Launch the slaves
   
   "${SPARK_HOME}/sbin/slaves.sh" cd "${SPARK_HOME}" \; "${SPARK_HOME}/sbin/start-slave.sh" "spark://$SPARK_MASTER_HOST:$SPARK_MASTER_PORT"

   會呼叫slaves.sh指令碼來啟動叢集中的Worker，其中cd後面的所有字串都是傳遞給slaves.sh的引數。

2. slaves.sh

   
   # slaves.sh
   
   for slave in `echo "$HOSTLIST"|sed  "s/#.*$//;/^$/d"`; do
    if [ -n "${SPARK_SSH_FOREGROUND}" ]; then
      ssh $SPARK_SSH_OPTS "$slave" $"${@// /\\ }" \
        2>&1 | sed "s/^/$slave: /"
    else
      ssh $SPARK_SSH_OPTS "$slave" $"${@// /\\ }" \
        2>&1 | sed "s/^/$slave: /" &
    fi
    if [ "$SPARK_SLAVE_SLEEP" != "" ]; then
      sleep $SPARK_SLAVE_SLEEP
    fi
   done

   在slaves.sh中，會解析slaves檔案，獲得當前叢集中的所有從節點host，最後會執行上面的程式碼段，意思是使用ssh連線到某個slave中，然後執行cd "${SPARK_HOME}" \; "${SPARK_HOME}/sbin/start-slave.sh" "spark://$SPARK_MASTER_HOST:$SPARK_MASTER_PORT"，這樣就成功的在遠端slave上啟動了一個Worker例項。

3. start-slave.sh

   該指令碼中又會通過spark-daemon.sh指令碼，來啟動org.apache.spark.worker.Worker類，spark-daemon.sh上面已經詳細介紹過了，只是啟動類不同，這裡不再複述了。

   "${SPARK_HOME}/sbin"/spark-daemon.sh start $CLASS $WORKER_NUM \
       --webui-port "$WEBUI_PORT" $PORT_FLAG $PORT_NUM $MASTER "[email protected]"

Worker的初始化

Worker的初始化比較簡單，就是向masterRpcAddresses列表中的Master註冊，如果Master是Active並且Worker沒有註冊過，那麼Master會回覆Worker訊息RegisteredWorker，表示Worker註冊成功；如果註冊失敗，那麼回覆RegisterWorkerFailed，Worker會退出。

Worker向Master註冊的時候有重試機制，即在指定時間如果收不到Master的響應，那麼Worker將會重新發送註冊請求。目前重新次數至多為16次。為了避免所有的Worker同時刻向Master傳送註冊請求，每次重試的時間間隔是隨機的，前6次的重試間隔在5~15s，而後10次的重試間隔在30~90秒。