Spark --- 启动、运行、关闭过程
2024-09-01 07:40:05
// scalastyle:off println
package org.apache.spark.examples import scala.math.random import org.apache.spark._ /** Computes an approximation to pi */
object SparkPi {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("Spark Pi")
val spark = new SparkContext(conf)
val slices = if (args.length > ) args().toInt else
val n = math.min(100000L * slices, Int.MaxValue).toInt // avoid overflow
val count = spark.parallelize( until n, slices).map { i =>
val x = random * -
val y = random * -
if (x*x + y*y < ) else
}.reduce(_ + _)
println("Pi is roughly " + 4.0 * count / n)
spark.stop()
}
}
[abc@search-engine---dev4 spark]$ ./bin/run-example SparkPi Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties // :: INFO SparkContext: Running Spark version 1.6. // :: WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable #进行acls用户权限认证 // :: INFO SecurityManager: Changing view acls to: abc // :: INFO SecurityManager: Changing modify acls to: abc // :: INFO SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(abc); users with modify permissions: Set(abc) // :: INFO Utils: Successfully started service 'sparkDriver' on port . // :: INFO Slf4jLogger: Slf4jLogger started #启动远程监听服务,端口是36739,Spark的通信工作由akka来实现 // :: INFO Remoting: Starting remoting // :: INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriverActorSystem@127.0.0.1:36739] // :: INFO Utils: Successfully started service 'sparkDriverActorSystem' on port . #注册MapOutputTracker,BlockManagerMaster,BlockManager // :: INFO SparkEnv: Registering MapOutputTracker // :: INFO SparkEnv: Registering BlockManagerMaster #分配存储空间,包括磁盘空间和内存空间 // :: INFO DiskBlockManager: Created local directory at /tmp/blockmgr-8a68c39e-40e5-43ca-b21e-081ef8d278e2 // :: INFO MemoryStore: MemoryStore started with capacity 511.1 MB // :: INFO SparkEnv: Registering OutputCommitCoordinator // :: INFO Utils: Successfully started service 'SparkUI' on port . // :: INFO SparkUI: Started SparkUI at http://127.0.0.1:4040 // :: INFO HttpFileServer: HTTP File server directory is /tmp/spark-3ef0b16c-fe81-482e--30571da062e7/httpd-796af3e2-122c---f4aa7d32bb04 #启动HTTP服务,可以通过界面查看服务和任务运行情况 // :: INFO HttpServer: Starting HTTP Server // :: INFO Utils: Successfully started service 'HTTP file server' on port . #启动SparkContext,并上传本地运行的jar包到http://127.0.0.1:54315 // :: INFO SparkContext: Added JAR file:/usr/local/spark/lib/spark-examples-1.6.-hadoop2.6.0.jar at http://127.0.0.1:54315/jars/spark-examples-1.6.1-hadoop2.6.0.jar with timestamp 1465285404966 // :: INFO Executor: Starting executor ID driver on host localhost // :: INFO Utils: Successfully started service 'org.apache.spark.network.netty.NettyBlockTransferService' on port . // :: INFO NettyBlockTransferService: Server created on // :: INFO BlockManagerMaster: Trying to register BlockManager // :: INFO BlockManagerMasterEndpoint: Registering block manager localhost: with 511.1 MB RAM, BlockManagerId(driver, localhost, ) // :: INFO BlockManagerMaster: Registered BlockManager #Spark提交了一个job给DAGScheduler // :: INFO SparkContext: Starting job: reduce at SparkPi.scala: #DAGScheduler收到一个编号为0的含有2个partitions分区的job // :: INFO DAGScheduler: Got job (reduce at SparkPi.scala:) with output partitions #将job转换为编号为0的stage // :: INFO DAGScheduler: Final stage: ResultStage (reduce at SparkPi.scala:) #DAGScheduler在submitting stage之前,首先寻找本次stage的parents,如果missing parents为空,则submitting stage; #如果有,会对parents stage进行递归submit stage,随之又将stage 0分成了2个task,提交给TaskScheduler的submitTasks方法。 #对于某些简单的job,如果它没有依赖关系,并且只有一个partition,这样的job会使用local thread处理而并不会提交到TaskScheduler上处理。 // :: INFO DAGScheduler: Parents of final stage: List() // :: INFO DAGScheduler: Missing parents: List() // :: INFO DAGScheduler: Submitting ResultStage (MapPartitionsRDD[] at map at SparkPi.scala:), which has no missing parents // :: INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 1904.0 B, free 1904.0 B) // :: INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 1218.0 B, free 3.0 KB) // :: INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost: (size: 1218.0 B, free: 511.1 MB) // :: INFO SparkContext: Created broadcast from broadcast at DAGScheduler.scala: // :: INFO DAGScheduler: Submitting missing tasks from ResultStage (MapPartitionsRDD[] at map at SparkPi.scala:) #TaskSchedulerImpl是TaskScheduler的实现类,接收了DAGScheduler提交的2个task // :: INFO TaskSchedulerImpl: Adding task set 0.0 with tasks // :: INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID , localhost, partition ,PROCESS_LOCAL, bytes) // :: INFO TaskSetManager: Starting task 1.0 in stage 0.0 (TID , localhost, partition ,PROCESS_LOCAL, bytes) #Executor接收任务后则从远程的服务器中将运行jar包存放到本地,然后进行计算,并各自汇报了任务执行状态 // :: INFO Executor: Running task 1.0 in stage 0.0 (TID ) // :: INFO Executor: Running task 0.0 in stage 0.0 (TID ) // :: INFO Executor: Fetching http://127.0.0.1:54315/jars/spark-examples-1.6.1-hadoop2.6.0.jar with timestamp 1465285404966 // :: INFO Utils: Fetching http://127.0.0.1:54315/jars/spark-examples-1.6.1-hadoop2.6.0.jar to /tmp/spark-3ef0b16c-fe81-482e-8446-30571da062e7/userFiles-b021b090-3024-421c-b4b0-73fc9f723f44/fetchFileTemp4760324069006875921.tmp // :: INFO Executor: Adding file:/tmp/spark-3ef0b16c-fe81-482e--30571da062e7/userFiles-b021b090--421c-b4b0-73fc9f723f44/spark-examples-1.6.-hadoop2.6.0.jar to class loader // :: INFO Executor: Finished task 1.0 in stage 0.0 (TID ). bytes result sent to driver // :: INFO Executor: Finished task 0.0 in stage 0.0 (TID ). bytes result sent to driver #TaskSetManager、SparkContent各自收到任务完成报告 // :: INFO TaskSetManager: Finished task 1.0 in stage 0.0 (TID ) in ms on localhost (/) // :: INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID ) in ms on localhost (/) // :: INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool // :: INFO DAGScheduler: ResultStage (reduce at SparkPi.scala:) finished in 2.217 s // :: INFO DAGScheduler: Job finished: reduce at SparkPi.scala:, took 2.877995 s #打印程序执行结果 Pi is roughly 3.14282 #Spark服务关闭 // :: INFO SparkUI: Stopped Spark web UI at http://127.0.0.1:4040 // :: INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped! // :: INFO MemoryStore: MemoryStore cleared // :: INFO BlockManager: BlockManager stopped // :: INFO BlockManagerMaster: BlockManagerMaster stopped // :: INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped! // :: INFO RemoteActorRefProvider$RemotingTerminator: Shutting down remote daemon. // :: INFO RemoteActorRefProvider$RemotingTerminator: Remote daemon shut down; proceeding with flushing remote transports. // :: INFO SparkContext: Successfully stopped SparkContext // :: INFO RemoteActorRefProvider$RemotingTerminator: Remoting shut down. // :: INFO ShutdownHookManager: Shutdown hook called // :: INFO ShutdownHookManager: Deleting directory /tmp/spark-3ef0b16c-fe81-482e--30571da062e7/httpd-796af3e2-122c---f4aa7d32bb04 // :: INFO ShutdownHookManager: Deleting directory /tmp/spark-3ef0b16c-fe81-482e--30571da062e7
最新文章
- 【DWR系列03】- DWR主要类详解
- 解析ThreadLocal
- [转]配置sonar、jenkins进行持续审查
- java程序员的技能要求
- 解决SSH会话连接超时问题
- JavaScript中的直接量与初始器的区别
- 用angular实时获取本地localStorage数据,实现一个模拟后台数据登入的效果
- ###《Machine Learning in Action》 - KNN
- 【Tools】Chrome开发者工具详解
- 基于Android Volley的网络请求工具
- Apache虚拟主机的配置
- Oracle存储过程 使用游标、数组的配合查询
- UI 事件处理
- UVA11404:Palindromic Subsequence
- pycharm上传代码到github
- Codeforces 1114 - A/B/C/D/E/F - (Undone)
- iOS的SVN
- SALALchemy Session与scoped_session的源码分析
- git排错
- 学习ML.NET(2): 使用模型进行预测
热门文章
- linux后台执行命令:&;和nohup
- DownloadProvider源码解析——与Volley对比
- 【typecho】解决使用分隔符 <;!--more-->;标签后首页文字下面出现一段空白
- linux下的文件解压命令
- 校验字符串是否是JSON格式,将不规则展示的json格式的字符串进行规则展示(json格式化)
- bitcoinj学习记录
- a buzzword to refer to modern Web technologies
- [knowledge][lisp] lisp与AI
- Flink &ndash; JobManager.submitJob
- jquery.axios无刷新机制删除