Scala進階之路-Spark底層通信小案例

　　　　　　　　　　　Scala進階之路-Spark底層通信小案例

　　　　　　　　　　　　　　　　　　　　　　　　　　　　作者：尹正傑

版權聲明：原創作品，謝絕轉載！否則將追究法律責任。

 

 

 

 

一.Spark Master和worker通信過程簡介

　　1>.Worker會向master注冊自己；

　　2>.Master收到worker的注冊信息之后，會告訴你已經注冊成功，並給worker發送啟動執行器的消息；

　　3>.Worker收到master的注冊消息之后，會定期向master匯報自己的狀態；

　　4>.master收到worker的心跳信息后，定期的更新worker的狀態，因為worker在發送心跳的時候會攜帶心跳發送的時間，master會監測master發送過來的心跳信時間和當前時間的差，如果大於5分鍾，master會監測發送過來的心跳時間和當前時間的差，如果大於5分鍾，則認為worker已死。然后master在分配任務的時候就不會給worker下發任務！

　　關於Master和Worker之間的通信機制，我們可以用以下一張圖介紹：

二.編寫源代碼

1>.Maven依賴

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.org.yinzhengjie</groupId>
    <artifactId>MyActor</artifactId>
    <version>1.0-SNAPSHOT</version>

    <!-- 定義一下常量 -->
    <properties>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.8</scala.version>
        <scala.compat.version>2.11</scala.compat.version>
        <akka.version>2.4.17</akka.version>
    </properties>

    <dependencies>
        <!-- 添加scala的依賴 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>

        <!-- 添加akka的actor依賴 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>

        <!-- 多進程之間的Actor通信 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-remote_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>
    </dependencies>

    <!-- 指定插件-->
    <build>
        <!-- 指定源碼包和測試包的位置 -->
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
    </build>
</project>

2>.MessageProtocol源代碼

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Scala%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:[e-mail]
*/
package cn.org.yinzhengjie.spark

/**
  * worker -> master ,即worker向master發送消息
  */
case class RegisterWorkerInfo(id: String, core: Int, ram: Int)          // worker向master注冊自己（信息）
case class HearBeat(id: String)                                        // worker給master發送心跳信息

/**
  * master -> worker，即master向worker發送消息
  */
case object RegisteredWorkerInfo                    // master向worker發送注冊成功消息
case object SendHeartBeat                           // worker 發送發送給自己的消息，告訴自己說要開始周期性的向master發送心跳消息
case object CheckTimeOutWorker                      //master自己給自己發送一個檢查超時worker的信息,並啟動一個調度器，周期新檢測刪除超時worker
case object RemoveTimeOutWorker                     // master發送給自己的消息，刪除超時的worker

/**
  * 定義存儲worker信息的類
  * @param id          :    每個worker的id是不變的且唯一的！
  * @param core        :    機器的核數
  * @param ram         :    內存大小
  */
case class WorkerInfo(val id: String, core: Int, ram: Int) {
    //定義最后一次的心跳時間,初始值為null。
    var lastHeartBeatTime: Long = _
}

3>.SparkWorker源代碼

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Scala%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:[e-mail]
*/
package cn.org.yinzhengjie.spark

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._          // 導入時間單位

/**
  * 定義主構造器，用於指定master的地址
  */
class SparkWorker(masterUrl: String) extends Actor{
    var masterProxy:ActorSelection = _                  //定義master的引用對象(actorRef)
    val workId:String = UUID.randomUUID().toString      //定義worker的uuid，每個worker的id是不變的且唯一的！
    /**
      * 通過preStart方法拿到master的引用對象(actorRef)，我們重寫該方法就會在receive方法執行之前執行！也就是拿到master對象只需要拿一次。
      */
    override def preStart(): Unit = {
        masterProxy = context.actorSelection(masterUrl)
    }
    override def receive: Receive = {
        case "started" => { // 自己已就緒
            // 向master注冊自己的信息，id, core, ram
            masterProxy ! RegisterWorkerInfo(workId, 4, 32 * 1024) // 此時master會收到該條信息
        }

        /**
          * 處理master發送給自己的注冊成功消息
          */
        case RegisteredWorkerInfo => {
            import context.dispatcher // 使用調度器時候必須導入dispatcher，因為該包涉及到隱式轉換的東西。
            /**
              *         worker通過"context.system.scheduler.schedule"啟動一個定時器，定時向master 發送心跳信息，需要指定
              * 四個參數：
              *         第一個參數是需要指定延時時間，此處指定的間隔時間為0毫秒；
              *         第二個參數是間隔時間，即指定定時器的周期性執行時間，我們這里指定為1秒；
              *         第三個參數是發送消息給誰，我們這里指定發送消息給自己，使用變量self即可；
              *         第四個參數是指發送消息的具體內容；
              *         注意：由於我們將消息周期性的發送給自己，因此我們自己需要接受消息並處理，也就是需要定義下面的SendHeartBeat
              */
            context.system.scheduler.schedule(0 millis, 1000 millis, self, SendHeartBeat)
        }
        case SendHeartBeat => {
            // 開始向master發送心跳了
            println(s"------- $workId 發送心跳 -------")
            masterProxy ! HearBeat(workId) // 此時master將會收到心跳信息
        }
    }
}


object SparkWorker {
    def main(args: Array[String]): Unit = {
        // 檢驗參數
        if(args.length != 4) {
            println(
                """
                  |請輸入參數：<host> <port> <workName> <masterURL>
                """.stripMargin)
            sys.exit() // 退出程序
        }
        /**
          * 定義參數，主機，端口號，worker名稱以及master的URL。
          */
        val host = args(0)
        val port = args(1)
        val workName = args(2)
        val masterURL = args(3)
        /**
          * 我們使用ConfigFactory.parseString來創建讀取參數配置的對象config
          */
        val config = ConfigFactory.parseString(
            s"""
              |akka.actor.provider="akka.remote.RemoteActorRefProvider"
              |akka.remote.netty.tcp.hostname=$host
              |akka.remote.netty.tcp.port=$port
            """.stripMargin)
        val actorSystem = ActorSystem("sparkWorker", config)
        /**
          * 創建worker的actorRef
          */
        val workerActorRef = actorSystem.actorOf(Props(new SparkWorker(masterURL)), workName)
        workerActorRef ! "started"      //給自己發送一個以啟動的消息，表示自己已經就緒了
    }
}

4>.SparkMaster源代碼

/*
@author :yinzhengjie
Blog:http://www.cnblogs.com/yinzhengjie/tag/Scala%E8%BF%9B%E9%98%B6%E4%B9%8B%E8%B7%AF/
EMAIL:[e-mail]
*/
package cn.org.yinzhengjie.spark

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._

class SparkMaster extends Actor{

    // 定義存儲worker的信息的saveWorkerInfo對象
    val saveWorkerInfo = collection.mutable.HashMap[String, WorkerInfo]()

//    override def preStart(): Unit = {
//        context.system.scheduler.schedule(0 millis, 6000 millis, self, RemoveTimeOutWorker)
//    }

    override def receive: Receive = {
        /**
          * 處理收到worker注冊過來的信息
          */
        case RegisterWorkerInfo(wkId, core, ram) => {
            /**
              * 存儲之前需要判斷之前是否已經存儲過了，如果沒有存儲就以wkId為key將worker的信息存儲起來，存儲到HashMap，
              */
            if (!saveWorkerInfo.contains(wkId)) {
                val workerInfo = new WorkerInfo(wkId, core, ram)
                saveWorkerInfo += ((wkId, workerInfo))
                /**
                  * master存儲完worker注冊的數據之后，要告訴worker說你已經注冊成功
                  */
                sender() ! RegisteredWorkerInfo // 此時worker會收到注冊成功消息
            }
        }
        /**
          *  master收到worker的心跳消息之后，更新woker的上一次心跳時間
          */
        case HearBeat(wkId) => {
            val workerInfo = saveWorkerInfo(wkId)
            val currentTime = System.currentTimeMillis()
            workerInfo.lastHeartBeatTime = currentTime          // 更改心跳時間
        }
        case CheckTimeOutWorker => {
            import context.dispatcher // 使用調度器時候必須導入dispatcher，因為該包涉及到隱式轉換的東西。
            context.system.scheduler.schedule(0 millis, 5000 millis, self, RemoveTimeOutWorker)
        }
        case RemoveTimeOutWorker => {
            /**
              *         將hashMap中的所有的value都拿出來，然后查看當前時間和上一次心跳時間差是否超過三次心跳時間，
              * 即三次沒有發送心跳信息就認為超時，每次心跳時間默認為1000毫秒，三次則為3000毫秒
              */
            val workerInfos = saveWorkerInfo.values
            val currentTime = System.currentTimeMillis()


            workerInfos
              .filter(workerInfo => currentTime - workerInfo.lastHeartBeatTime > 3000)        //過濾超時的worker
              .foreach(workerInfo => saveWorkerInfo.remove(workerInfo.id))                    //將過濾超時的worker刪除掉
            println(s"====== 還剩 ${saveWorkerInfo.size} 存活的Worker ======")
        }
    }
}

object SparkMaster {
    private var name = ""
    private val age = 100
    def main(args: Array[String]): Unit = {
        // 檢驗參數
        if(args.length != 3) {
            println(
                """
                  |請輸入參數：<host> <port> <masterName>
                """.stripMargin)
            sys.exit() // 退出程序
        }
        /**
          * 定義參數，主機，端口號，master名稱
          */
        val host = args(0)
        val port = args(1)
        val masterName = args(2)
        /**
          * 我們使用ConfigFactory.parseString來創建讀取參數配置的對象config
          */
        val config = ConfigFactory.parseString(
            s"""
               |akka.actor.provider="akka.remote.RemoteActorRefProvider"
               |akka.remote.netty.tcp.hostname=$host
               |akka.remote.netty.tcp.port=$port
            """.stripMargin)

        val actorSystem = ActorSystem("sparkMaster", config)
        val masterActorRef = actorSystem.actorOf(Props[SparkMaster], masterName)
        /**
          * 自己給自己發送一個消息，去啟動一個調度器，定期的檢測HashMap中超時的worker
          */
        masterActorRef ! CheckTimeOutWorker
    }
}

 

三.本機測試

1>.啟動master端

配置參數如下：
127.0.0.1 8888 master

2>.啟動woker端

兩個worker的配置參數如下:
127.0.0.1 6665 worker akka.tcp://sparkMaster@127.0.0.1:8888//user/master
127.0.0.1 6666 worker akka.tcp://sparkMaster@127.0.0.1:8888//user/master

　　服務端輸出如下：

 

四.master worker打包部署到linux多台服務測試

1>.打包SparkMaster

　　第一步：修改Maven配置如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.org.yinzhengjie</groupId>
    <artifactId>MyActor</artifactId>
    <version>1.0-SNAPSHOT</version>
    <!-- 定義一下常量 -->
    <properties>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.8</scala.version>
        <scala.compat.version>2.11</scala.compat.version>
        <akka.version>2.4.17</akka.version>
    </properties>
    <dependencies>
        <!-- 添加scala的依賴 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <!-- 添加akka的actor依賴 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>
        <!-- 多進程之間的Actor通信 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-remote_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>
    </dependencies>
    <!-- 指定插件-->
    <build>
        <!-- 指定源碼包和測試包的位置 -->
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <!-- 指定編譯scala的插件 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!-- maven打包的插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>reference.conf</resource>
                                </transformer>
                                <!-- 指定main方法:cn.org.yinzhengjie.spark.SparkMaster -->
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>cn.org.yinzhengjie.spark.SparkMaster</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

　　第二步：點擊package開始打包：

　　第三步：查看依賴包內部結構：

 

2>.打包SparkWorker

　　第一步：修改Maven配置如下：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.org.yinzhengjie</groupId>
    <artifactId>MyActor</artifactId>
    <version>1.0-SNAPSHOT</version>
    <!-- 定義一下常量 -->
    <properties>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.8</scala.version>
        <scala.compat.version>2.11</scala.compat.version>
        <akka.version>2.4.17</akka.version>
    </properties>
    <dependencies>
        <!-- 添加scala的依賴 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>
        <!-- 添加akka的actor依賴 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-actor_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>
        <!-- 多進程之間的Actor通信 -->
        <dependency>
            <groupId>com.typesafe.akka</groupId>
            <artifactId>akka-remote_${scala.compat.version}</artifactId>
            <version>${akka.version}</version>
        </dependency>
    </dependencies>
    <!-- 指定插件-->
    <build>
        <!-- 指定源碼包和測試包的位置 -->
        <sourceDirectory>src/main/scala</sourceDirectory>
        <testSourceDirectory>src/test/scala</testSourceDirectory>
        <plugins>
            <!-- 指定編譯scala的插件 -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!-- maven打包的插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                    <resource>reference.conf</resource>
                                </transformer>
                                <!-- 指定main方法:cn.org.yinzhengjie.spark.SparkWorker -->
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>cn.org.yinzhengjie.spark.SparkWorker</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

　　接下來的兩步還是和上面的步驟一直，將打包完成后的文件改名並查看主類信息如下：

3>.開啟三台虛擬機並在master節點上傳master.jar並運行

[yinzhengjie@s101 download]$ ll
total 67320
-rw-r--r--. 1 yinzhengjie yinzhengjie 20124547 Jul 31 20:42 master.jar
-rw-r--r--. 1 yinzhengjie yinzhengjie 28678231 Jul 20 21:18 scala-2.11.8.tgz
-rw-r--r--. 1 yinzhengjie yinzhengjie 20124541 Jul 31 21:52 worker.jar
[yinzhengjie@s101 download]$ 
[yinzhengjie@s101 download]$ 
[yinzhengjie@s101 download]$ java -jar master.jar 172.16.30.101 8888 master

 

4>.將worker.jar包上傳到另外的兩個節點並運行，如下:

　　172.16.30.102節點操作如下：
[yinzhengjie@s102 download]$ ll
total 19656
-rw-r--r--. 1 yinzhengjie yinzhengjie 20124541 Jul 31 22:01 worker.jar
[yinzhengjie@s102 download]$ 
[yinzhengjie@s102 download]$ java -jar worker.jar 172.16.30.102 6665 worker akka.tcp://sparkMaster@172.16.30.101:8888//user/master
[yinzhengjie@s102 download]$ 

　　172.16.30.103節點操作如下：
[yinzhengjie@s103 download]$ ll
total 19656
-rw-r--r--. 1 yinzhengjie yinzhengjie 20124541 Jul 31 22:00 worker.jar
[yinzhengjie@s103 download]$ 
[yinzhengjie@s103 download]$ 
[yinzhengjie@s103 download]$ java -jar worker.jar 172.16.30.103 6665 worker akka.tcp://sparkMaster@172.16.30.101:8888//user/master

　　172.16.30.102節點操作如下：

 

　　172.16.30.103節點操作如下：

　　172.16.30.101輸出信息如下: