spark源碼之standalone模式下master持久化引擎講解

1、Spark 源碼之 Standalone 模式下 master 持久化引擎講解Standalone 模式下 Master 為了保證故障恢復(fù)，會(huì)持久化一些重要的數(shù)據(jù)，來避免 master故障導(dǎo)致集群不可用這種情況(也即單點(diǎn)故障)。目前，有四種持久化策略: 1,基于 zookeeper 的持久化引擎。2,基于文件的持久化引擎。3,用戶自定義持久化引擎。4,不使用持久化引擎。在 master 的 OnStart 方法中，對應(yīng)的源碼如下:val serializer =new JavaSerializer(conf)val (persistenceEngine_, leaderElectionAgen

2、t_) = RECOVERY_MODE matchcase ZOOKEEPER =logInfo(Persisting recovery sval zkFactory =e to ZooKeeper)new ZooKeeperRecoveryModeFactory(conf, serializer)(zkFactory.createPersistenceEngine(),zkFactory.creaeaderElectionAgent(this)case FILESYSTEM =val fsFactory =new FileSystemRecoveryModeFactory(conf, ser

3、ializer) (fsFactory.createPersistenceEngine(),fsFactory.creaeaderElectionAgent(this) case CUSTOM =val clazz = Utils.classForName(conf.get(spark.deploy.recoveryMode.factory) val factory = clazz.getConstructor(classOfSparkConf, classOfSerializer).newInstance(conf, serializer).asInstanceOfStandaloneRec

4、overyModeFactory(factory.createPersistenceEngine(), factory.creacase _ =eaderElectionAgent(this)(new BlackHolePersistenceEngine(), new MonarchyLeaderAgent(this)persistenceEngine = persistenceEngine_leaderElectionAgent = leaderElectionAgent_默認(rèn)，情況下是無持久化引擎，也就是沒有 ha 策略。Spark 提供的可用的 ha 策略：基于文件系統(tǒng)的和基于 zook

5、eeper。配置方法如下:基于文件系統(tǒng)：基于 zookeeper:生產(chǎn)環(huán)境中可用的是基于 zookeeper 的持久化引擎。propertyMeaningspark.deploy.recoveryModeZOOKEEPERspark.deploy.zookeeper.urle.g., 192.168.1.100:2181,192.168.1.101:2181spark.deploy.zookeeper.dirzookeeper 保存恢復(fù)狀態(tài)的目錄propertyMeaningspark.deploy.recoveryModeFILESYSTEMspark.deploy.recoveryDire

6、ctory用來恢復(fù)狀態(tài)的目錄基于 zookeeper 持久化策略，會(huì)允許最終是一個(gè) leader，其余是 standby。同時(shí)運(yùn)行多個(gè) master，然后支持 leader，Spark 的 Master 的 leader實(shí)現(xiàn)Spark 源碼里面使用的是 CuratorFramework，跟 zookeeper 交流。該框架有以下特點(diǎn):1，自動(dòng)連接管理：自動(dòng)處理 zookeeper 的連接和重試存在一些潛在；可以 watchNodeDhanged event 和獲取 updateServerList;Watches 可以自動(dòng)被 Cruator recipes 刪除；2，更加簡潔的 API：簡化

7、raw zookeeper 方法，事件等；提供現(xiàn)代流式 API 接口3，Recipe 實(shí)現(xiàn)：leader，分布式鎖，path 緩存，和 watcher,分布式隊(duì)列，Barriers等。Spark 源碼里面使用了 LeaderLatch 實(shí)現(xiàn)點(diǎn)類型來做，zookeeper 有四種節(jié)點(diǎn)類型:1，持久節(jié)點(diǎn)（PERSISTENT）功能。這個(gè)實(shí)現(xiàn)實(shí)際是基于 zookeeper 的節(jié)節(jié)點(diǎn)創(chuàng)建后，會(huì)一直存在，不會(huì)因客戶端會(huì)話失效而刪除；2，持久順序節(jié)點(diǎn)（PERSISTENT_SEQUENTIAL）基本特性與持久節(jié)點(diǎn)一致，創(chuàng)建節(jié)點(diǎn)的過程中，zookeeper 會(huì)在其名字后自動(dòng)追加一個(gè)單調(diào)增長的數(shù)字后綴，作為新

8、的節(jié)點(diǎn)名；3，臨時(shí)節(jié)點(diǎn)（EPHEMERAL）客戶端會(huì)話失效或連接關(guān)閉后，該節(jié)點(diǎn)會(huì)被自動(dòng)刪除，且不能再臨時(shí)節(jié)點(diǎn)下面創(chuàng)建子節(jié)點(diǎn)。4，臨時(shí)順序節(jié)點(diǎn)（EPHEMERAL_SEQUENTIAL）基本特性與臨時(shí)節(jié)點(diǎn)一致，創(chuàng)建節(jié)點(diǎn)的過程中，zookeeper 會(huì)在其名字后自動(dòng)追加一個(gè)單調(diào)增長的數(shù)字后綴，作為新的節(jié)點(diǎn)名；LeaderLatch 實(shí)現(xiàn) leader實(shí)際上基于臨時(shí)順序節(jié)點(diǎn)來做的。Spark 源碼里面基于 zookeeper 的 leader在 master 的 OnStart 方法里面具體實(shí)現(xiàn)過程源碼如下:實(shí)際是在構(gòu)建 zookeeper 的持久化引擎的時(shí)候，構(gòu)建的在 creaeaderElect

9、ionAgent 方法里面構(gòu)建了new ZooKeeperLeaderElectionAgent(master, conf) 該對象，繼承了 LeaderLatchListener，并且覆蓋了 notLeader 和 isLeader 兩個(gè)重要的方法具體。在 ZooKeeperLeaderElectionAgent 構(gòu)建的時(shí)候調(diào)用了自己的 start 方法，該方法構(gòu)建了LeaderLatch，并添加 ZooKeeperLeaderElectionAgent 作為其 listener。Leader在 zookeeper 的臨時(shí)節(jié)點(diǎn)的路徑為private def start() logInfo(S

10、tarting ZooKeeper LeaderElection agent)zk = SparkCuratorUtil.newnt(conf) leaderLatch = new LeaderLatch(zk, WORKING_DIR) leaderLatch.addListener(this) leaderLatch.start()(zkFactory.createPersistenceEngine(), zkFactory.creaeaderElectionAgent(this)leaderElectionAgent =leaderElectionAgent_浪尖技術(shù)224209501執(zhí)

11、行 ZooKeeperLeaderElectionAgent 對象的 start 方法之后，每當(dāng)該對象所在的 master由 standby 變?yōu)?Leader 的時(shí)候，會(huì)調(diào)用 isLeader()方法。由 Leader 變?yōu)?StandBy 的時(shí)候會(huì)調(diào)用 notLeader()。就可以在這兩個(gè)方法里實(shí)現(xiàn)自己要的狀態(tài)切換的相關(guān)操作。要實(shí)現(xiàn)，自己應(yīng)用的 ha，也可基于此方法。浪尖技術(shù)224209501override def isLeader() synchronized / could have lost leadership by now. if (!leaderLatch.hasLeadership) returnlogInfo(We have gained leadership) updaeadershipSus(true)override def notLeader() synchronized / could have gained leadership b