- 論壇徽章:
- 10
|
1. 有人說Spark就是內(nèi)存版的MapReduce，對(duì)此你怎么看？
(1)spark與以MapReduce為出發(fā)點(diǎn)的hadoop有幾乎同樣的效果，絕不是同樣的處理方式。以下簡(jiǎn)單介紹不同之處。
Spark 是一種與 Hadoop 相似的開源集群計(jì)算環(huán)境，但是兩者之間還存在一些不同之處，這些有用的不同之處使 Spark 在某些工作負(fù)載方面表現(xiàn)得更加優(yōu)越，換句話說，Spark 啟用了內(nèi)存分布數(shù)據(jù)集，除了能夠提供交互式查詢外，它還可以優(yōu)化迭代工作負(fù)載。
Spark 是在 Scala 語言中實(shí)現(xiàn)的，它將 Scala 用作其應(yīng)用程序框架。與 Hadoop 不同，Spark 和 Scala 能夠緊密集成，其中的 Scala 可以像操作本地集合對(duì)象一樣輕松地操作分布式數(shù)據(jù)集。
盡管創(chuàng)建 Spark 是為了支持分布式數(shù)據(jù)集上的迭代作業(yè)，但是實(shí)際上它是對(duì) Hadoop 的補(bǔ)充，可以在 Hadoop 文件系統(tǒng)中并行運(yùn)行。通過名為 Mesos 的第三方集群框架可以支持此行為。Spark 由加州大學(xué)伯克利分校 AMP 實(shí)驗(yàn)室 (Algorithms, Machines, and People Lab) 開發(fā)，可用來構(gòu)建大型的、低延遲的數(shù)據(jù)分析應(yīng)用程序。
Screenshot-4.png (364.78 KB, 下載次數(shù): 386)
下載附件
2015-10-30 20:43 上傳
(2) 對(duì)于spark的部署，請(qǐng)參見我的博文：http://blog.chinaunix.net/uid-7374279-id-5200921.html
以下是部分介紹：
先從比較簡(jiǎn)單的說起，所謂的沒有ha是指master節(jié)點(diǎn)沒有ha。
組成cluster的兩大元素即Master和Worker。slave worker可以有1到多個(gè)，這些worker都處于active狀態(tài)。
Driver Application可以運(yùn)行在Cluster之內(nèi)，也可以在cluster之外運(yùn)行，先從簡(jiǎn)單的講起即Driver Application獨(dú)立于Cluster。那么這樣的整體框架如下圖所示，由driver，master和多個(gè)slave worker來共同組成整個(gè)的運(yùn)行環(huán)境。
執(zhí)行順序
步驟1 運(yùn)行master
$SPARK_HOME/sbin/start_master.sh
在 start_master.sh 中最關(guān)鍵的一句就是
"$sbin"/spark-daemon.sh start org.apache.spark.deploy.master.Master 1 --ip $SPARK_MASTER_IP --port $SPARK_MASTER_PORT --webui-port $SPARK_MASTER_WEBUI_PORT
檢測(cè)Master的jvm進(jìn)程
root 23438 1 67 22:57 pts/0 00:00:05 /opt/java/bin/java -cp :/root/working/spark-0.9.1-bin-hadoop2/conf:/root/working/spark-0.9.1-bin-hadoop2/assembly/target/scala-2.10/spark-assembly_2.10-0.9.1-hadoop2.2.0.jar -Dspark.akka.logLifecycleEvents=true -Djava.library.path= -Xms512m -Xmx512m org.apache.spark.deploy.master.Master --ip localhost --port 7077 --webui-port 8080
Master的日志在$SPARK_HOME/logs目錄下
步驟2 運(yùn)行worker，可以啟動(dòng)多個(gè)
./bin/spark-class org.apache.spark.deploy.worker.Worker spark://localhost:7077
worker運(yùn)行時(shí)，需要注冊(cè)到指定的master url，這里就是spark://localhost:7077.
Master側(cè)收到RegisterWorker通知，其處理代碼如下
case RegisterWorker(id, workerHost, workerPort, cores, memory, workerUiPort, publicAddress) =>
{
logInfo("Registering worker %s:%d with %d cores, %s RAM".format(
workerHost, workerPort, cores, Utils.megabytesToString(memory))) if (state == RecoveryState.STANDBY) { // ignore, don't send response } else if (idToWorker.contains(id)) {
sender ! RegisterWorkerFailed("Duplicate worker ID"
} else {
val worker = new WorkerInfo(id, workerHost, workerPort, cores, memory,
sender, workerUiPort, publicAddress) if (registerWorker(worker)) {
persistenceEngine.addWorker(worker)
sender ! RegisteredWorker(masterUrl, masterWebUiUrl)
schedule()
} else {
val workerAddress = worker.actor.path.address
logWarning("Worker registration failed. Attempted to re-register worker at same " + "address: " + workerAddress)
sender ! RegisterWorkerFailed("Attempted to re-register worker at same address: " + workerAddress)
}
}
}
步驟3 運(yùn)行Spark-shell
MASTER=spark://localhost:7077 $SPARK_HOME/bin/spark-shell
spark-shell屬于application，有關(guān)appliation的運(yùn)行日志存儲(chǔ)在 $SPARK_HOME/works 目錄下
spark-shell作為application,在Master側(cè)其處理的分支是RegisterApplication,具體處理代碼如下。
case RegisterApplication(description) => { if (state == RecoveryState.STANDBY) { // ignore, don't send response } else {
logInfo("Registering app " + description.name)
val app = createApplication(description, sender)
registerApplication(app)
logInfo("Registered app " + description.name + " with ID " + app.id)
persistenceEngine.addApplication(app)
sender ! RegisteredApplication(app.id, masterUrl)
schedule()
}
}
每當(dāng)有新的application注冊(cè)到master，master都要調(diào)度schedule函數(shù)將application發(fā)送到相應(yīng)的 worker，在對(duì)應(yīng)的worker啟動(dòng)相應(yīng)的ExecutorBackend. 具體代碼請(qǐng)參考Master.scala中的schedule函數(shù)，代碼就不再列出。
步驟4 結(jié)果檢測(cè)
/opt/java/bin/java -cp :/root/working/spark-0.9.1-bin-hadoop2/conf:/root/working/spark-0.9.1-bin-hadoop2/assembly/target/scala-2.10/spark-assembly_2.10-0.9.1-hadoop2.2.0.jar -Dspark.akka.logLifecycleEvents=true -Djava.library.path= -Xms512m -Xmx512m org.apache.spark.deploy.master.Master --ip localhost --port 7077 --webui-port 8080 root 23752 23745 21 23:00 pts/0 00:00:25 /opt/java/bin/java -cp :/root/working/spark-0.9.1-bin-hadoop2/conf:/root/working/spark-0.9.1-bin-hadoop2/assembly/target/scala-2.10/spark-assembly_2.10-0.9.1-hadoop2.2.0.jar -Djava.library.path= -Xms512m -Xmx512m org.apache.spark.repl.Main
root 23986 23938 25 23:02 pts/2 00:00:03 /opt/java/bin/java -cp :/root/working/spark-0.9.1-bin-hadoop2/conf:/root/working/spark-0.9.1-bin-hadoop2/assembly/target/scala-2.10/spark-assembly_2.10-0.9.1-hadoop2.2.0.jar -Dspark.akka.logLifecycleEvents=true -Djava.library.path= -Xms512m -Xmx512m org.apache.spark.deploy.worker.Worker spark://localhost:7077 root 24047 23986 34 23:02 pts/2 00:00:04 /opt/java/bin/java -cp :/root/working/spark-0.9.1-bin-hadoop2/conf:/root/working/spark-0.9.1-bin-hadoop2/assembly/target/scala-2.10/spark-assembly_2.10-0.9.1-hadoop2.2.0.jar -Xms512M -Xmx512M org.apache.spark.executor.CoarseGrainedExecutorBackend akka.tcp://spark@localhost:40053/user/CoarseGrainedScheduler 0 localhost 4 akka.tcp://sparkWorker@localhost:53568/user/Worker app-20140511230059-0000
從運(yùn)行的進(jìn)程之間的關(guān)系可以看出，worker和master之間的連接建立完畢之后，如果有新的driver application連接上master，master會(huì)要求worker啟動(dòng)相應(yīng)的ExecutorBackend進(jìn)程。此后若有什么Task需要運(yùn) 行，則會(huì)運(yùn)行在這些Executor之上�？梢詮囊韵碌娜罩拘畔⒌贸龃私Y(jié)論，當(dāng)然看源碼亦可。
14/05/11 23:02:36 INFO Worker: Asked to launch executor app-20140511230059-0000/0 for Spark shell 14/05/11 23:02:36 INFO ExecutorRunner: Launch command: "/opt/java/bin/java" "-cp" ":/root/working/spark-0.9.1-bin-hadoop2/conf:/root/working/spark-0.9.1-bin-hadoop2/assembly/target/scala-2.10/spark-assembly_2.10-0.9.1-hadoop2.2.0.jar" "-Xms512M" "-Xmx512M" "org.apache.spark.executor.CoarseGrainedExecutorBackend" "akka.tcp://spark@localhost:40053/user/CoarseGrainedScheduler" "0" "localhost" "4" "akka.tcp://sparkWorker@localhost:53568/user/Worker" "app-20140511230059-0000"
worker中啟動(dòng)exectuor的相關(guān)源碼見worker中的receive函數(shù)，相關(guān)代碼如下
case LaunchExecutor(masterUrl, appId, execId, appDesc, cores_, memory_) => if (masterUrl != activeMasterUrl) {
logWarning("Invalid Master (" + masterUrl + " attempted to launch executor."
} else { try {
logInfo("Asked to launch executor %s/%d for %s".format(appId, execId, appDesc.name))
val manager = new ExecutorRunner(appId, execId, appDesc, cores_, memory_, self, workerId, host,
appDesc.sparkHome.map(userSparkHome => new File(userSparkHome)).getOrElse(sparkHome),
workDir, akkaUrl, ExecutorState.RUNNING)
executors(appId + "/" + execId) = manager
manager.start()
coresUsed += cores_
memoryUsed += memory_
masterLock.synchronized {
master ! ExecutorStateChanged(appId, execId, manager.state, None, None)
}
} catch { case e: Exception => {
logError("Failed to launch exector %s/%d for %s".format(appId, execId, appDesc.name)) if (executors.contains(appId + "/" + execId)) {
executors(appId + "/" + execId).kill()
executors -= appId + "/" + execId
}
masterLock.synchronized {
master ! ExecutorStateChanged(appId, execId, ExecutorState.FAILED, None, None)
}
}
}
}
關(guān)于standalone的部署，需要詳細(xì)研究的源碼文件如下所列。
deploy/master/Master.scala
deploy/worker/worker.scala
executor/CoarseGrainedExecutorBackend.scala
查看進(jìn)程之間的父子關(guān)系，請(qǐng)用 "pstree"
使用下圖來小結(jié)單Master的部署情況。
類的動(dòng)態(tài)加載和反射
在談部署Driver到Cluster上之前，我們先回顧一下java的一大特性“類的動(dòng)態(tài)加載和反射機(jī)制”。本人不是一直寫java代碼出身，所以好多東西都是邊用邊學(xué)，難免掛一漏萬。
所謂的反射，其實(shí)就是要解決在運(yùn)行期實(shí)現(xiàn)類的動(dòng)態(tài)加載。
來個(gè)簡(jiǎn)單的例子
package test; public class Demo { public Demo() { System.out.println("Hi!"; } @SuppressWarnings("unchecked" public static void main(String[] args) throws Exception { Class clazz = Class.forName("test.Demo"; Demo demo = (Demo) clazz.newInstance(); }
}
談到這里，就自然想到了一個(gè)面試題，“談一談Class.forName和ClassLoader.loadClass的區(qū)別"。說到面試，我總是很沒有信心，面試官都很屌的,  。
在cluster中運(yùn)行Driver Application
上一節(jié)之所以寫到類的動(dòng)態(tài)加載與反射都是為了談這一節(jié)的內(nèi)容奠定基礎(chǔ)。
將Driver application部署到Cluster中，啟動(dòng)的時(shí)序大體如下圖所示。
首先啟動(dòng)Master,然后啟動(dòng)Worker
使用”deploy.Client"將Driver Application提交到Cluster中
./bin/spark-class org.apache.spark.deploy.Client launch [client-options] \
\
[application-options]
Master在收到RegisterDriver的請(qǐng)求之后，會(huì)發(fā)送LaunchDriver給worker,要求worker啟動(dòng)一個(gè)Driver的jvm process
Driver Application在新生成的JVM進(jìn)程中運(yùn)行開始時(shí)會(huì)注冊(cè)到master中，發(fā)送RegisterApplication給Master
Master發(fā)送LaunchExecutor給Worker，要求Worker啟動(dòng)執(zhí)行ExecutorBackend的JVM Process
一當(dāng)ExecutorBackend啟動(dòng)完畢，Driver Application就可以將任務(wù)提交到ExecutorBackend上面執(zhí)行，即LaunchTask指令
提交側(cè)的代碼，詳見deploy/Client.scala
driverArgs.cmd match { case "launch" => // TODO: We could add an env variable here and intercept it in `sc.addJar` that would // truncate filesystem paths similar to what YARN does. For now, we just require // people call `addJar` assuming the jar is in the same directory. val env = Map[String, String]()
System.getenv().foreach{case (k, v) => env(k) = v}
val mainClass = "org.apache.spark.deploy.worker.DriverWrapper" val classPathConf = "spark.driver.extraClassPath" val classPathEntries = sys.props.get(classPathConf).toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}
val libraryPathConf = "spark.driver.extraLibraryPath" val libraryPathEntries = sys.props.get(libraryPathConf).toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}
val javaOptionsConf = "spark.driver.extraJavaOptions" val javaOpts = sys.props.get(javaOptionsConf)
val command = new Command(mainClass, Seq("{{WORKER_URL}}", driverArgs.mainClass) ++
driverArgs.driverOptions, env, classPathEntries, libraryPathEntries, javaOpts)
val driverDescription = new DriverDescription(
driverArgs.jarUrl,
driverArgs.memory,
driverArgs.cores,
driverArgs.supervise,
command)
masterActor ! RequestSubmitDriver(driverDescription)
接收側(cè)
從Deploy.client發(fā)送出來的消息被誰接收呢？答案比較明顯，那就是Master。 Master.scala中的receive函數(shù)有專門針對(duì)RequestSubmitDriver的處理，具體代碼如下
case RequestSubmitDriver(description) => { if (state != RecoveryState.ALIVE) {
val msg = s"Can only accept driver submissions in ALIVE state. Current state: $state." sender ! SubmitDriverResponse(false, None, msg)
} else {
logInfo("Driver submitted " + description.command.mainClass)
val driver = createDriver(description)
persistenceEngine.addDriver(driver)
waitingDrivers += driver
drivers.add(driver)
schedule() // TODO: It might be good to instead have the submission client poll the master to determine // the current status of the driver. For now it's simply "fire and forget". sender ! SubmitDriverResponse(true, Some(driver.id),
s"Driver successfully submitted as ${driver.id}"
}
}
SparkEnv
SparkEnv對(duì)于整個(gè)Spark的任務(wù)來說非常關(guān)鍵，不同的role在創(chuàng)建SparkEnv時(shí)傳入的參數(shù)是不相同的，如Driver和Executor則存在重要區(qū)別。
在Executor.scala中，創(chuàng)建SparkEnv的代碼如下所示
private val env = { if (!isLocal) {
val _env = SparkEnv.create(conf, executorId, slaveHostname, 0,
isDriver = false, isLocal = false)
SparkEnv.set(_env)
_env.metricsSystem.registerSource(executorSource)
_env
} else {
SparkEnv.get }
}
Driver Application則會(huì)創(chuàng)建SparkContext，在SparkContext創(chuàng)建過程中，比較重要的一步就是生成SparkEnv,其代碼如下
private[spark] val env = SparkEnv.create( conf, "", conf.get("spark.driver.host", conf.get("spark.driver.port".toInt, isDriver = true, isLocal = isLocal, listenerBus = listenerBus)
SparkEnv.set(env)
Standalone模式下HA的實(shí)現(xiàn)
Spark在standalone模式下利用zookeeper來實(shí)現(xiàn)了HA機(jī)制，這里所說的HA是專門針對(duì)Master節(jié)點(diǎn)的，因?yàn)樯厦嫠械姆治隹梢钥闯鯩aster是整個(gè)cluster中唯一可能出現(xiàn)單點(diǎn)失效的節(jié)點(diǎn)。
采用zookeeper之后，整個(gè)cluster的組成如下圖所示。
為了使用zookeeper，Master在啟動(dòng)的時(shí)候需要指定如下的參數(shù)，修改conf/spark-env.sh, SPARK_DAEMON_JAVA_OPTS中添加如下選項(xiàng)。
System property Meaning
spark.deploy.recoveryMode Set to ZOOKEEPER to enable standby Master recovery mode (default: NONE).
spark.deploy.zookeeper.url The ZooKeeper cluster url (e.g., 192.168.1.100:2181,192.168.1.101:2181).
spark.deploy.zookeeper.dir The directory in ZooKeeper to store recovery state (default: /spark).
實(shí)現(xiàn)HA的原理
zookeeper提供了一個(gè)Leader Election機(jī)制，利用這個(gè)機(jī)制，可以實(shí)現(xiàn)HA功能，具體請(qǐng)參考 zookeeper recipes
在Spark中沒有直接使用zookeeper的api，而是使用了 curator ，curator對(duì)zookeeper做了相應(yīng)的封裝，在使用上更為友好。
小結(jié)
步步演進(jìn)講到在standalone模式下，如何利用zookeeper來實(shí)現(xiàn)ha。從中可以看出standalone master一個(gè)最主要的任務(wù)就是resource management和job scheduling，看到這兩個(gè)主要功能的時(shí)候，您也許會(huì)想到這不就是YARN要解決的問題。對(duì)了，從本質(zhì)上來說standalone是yarn的一個(gè) 簡(jiǎn)化版本。
2. 有人說Spark將來會(huì)替代Hadoop，你又怎么看？
未來會(huì)的，但現(xiàn)在還早。hadoop有些笨重，但歷史悠久。許多企業(yè)投入精力不會(huì)輕易放棄的。
|
|
免費(fèi)注冊(cè)
發(fā)表于 2015-10-28 21:26
）
