本帖最后由 pig2 于 2015-1-6 14:16 编辑 问题导读 １．Spark基于Akka来进行消息交互，那如何知道谁是接收方呢？ ２．对代码作了修改之后，如果并不想提交代码，那该如何将最新的内容同步到本地呢？ 概要 今天不谈Spark中什么复杂的技术实现，只稍为聊聊如何进行代码跟读。众所周知，Spark使用scala进行开发，由于scala有众多的语法糖，很多时候代码跟着跟着就觉着线索跟丢掉了，另外Spark基于Akka来进行消息交互，那如何知道谁是接收方呢？
new Throwable().printStackTrace 代码跟读的时候，经常会借助于日志，针对日志中输出的每一句，我们都很想知道它们的调用者是谁。但有时苦于对spark系统的了解程度不深，或者对scala认识不够，一时半会之内无法找到答案，那么有没有什么简便的办法呢？
我的办法就是在日志出现的地方加入下面一句话
new Throwable().printStackTrace() 复制代码 现在举一个实际的例子来说明问题。
比如我们在启动spark-shell之后，输入一句非常简单的sc.textFile("README.md")，会输出下述的log
14/07/05 19:53:27 INFO MemoryStore: ensureFreeSpace(32816) called with curMem=0, maxMem=308910489
14/07/05 19:53:27 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 32.0 KB, free 294.6 MB)
14/07/05 19:53:27 DEBUG BlockManager: Put block broadcast_0 locally took 78 ms
14/07/05 19:53:27 DEBUG BlockManager: Putting block broadcast_0 without replication took 79 ms
res0: org.apache.spark.rdd.RDD[String] = README.md MappedRDD[1] at textFile at :13
复制代码 那我很想知道是第二句日志所在的tryToPut函数是被谁调用的该怎么办？
办法就是打开MemoryStore.scala，找到下述语句
logInfo("Block %s stored as %s in memory (estimated size %s, free %s)".format(
blockId, valuesOrBytes, Utils.bytesToString(size), Utils.bytesToString(freeMemory)))
复制代码 new Throwable().printStackTrace() 复制代码 复制代码 14/07/05 19:53:27 INFO MemoryStore: ensureFreeSpace(32816) called with curMem=0, maxMem=308910489
14/07/05 19:53:27 WARN MemoryStore: just show the calltrace by entering some modified code
java.lang.Throwable
at org.apache.spark.storage.MemoryStore.tryToPut(MemoryStore.scala:182)
at org.apache.spark.storage.MemoryStore.putValues(MemoryStore.scala:76)
at org.apache.spark.storage.MemoryStore.putValues(MemoryStore.scala:92)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:699)
at org.apache.spark.storage.BlockManager.put(BlockManager.scala:570)
at org.apache.spark.storage.BlockManager.putSingle(BlockManager.scala:821)
at org.apache.spark.broadcast.HttpBroadcast.(HttpBroadcast.scala:52)
at org.apache.spark.broadcast.HttpBroadcastFactory.newBroadcast(HttpBroadcastFactory.scala:35)
at org.apache.spark.broadcast.HttpBroadcastFactory.newBroadcast(HttpBroadcastFactory.scala:29)
at org.apache.spark.broadcast.BroadcastManager.newBroadcast(BroadcastManager.scala:62)
at org.apache.spark.SparkContext.broadcast(SparkContext.scala:787)
at org.apache.spark.SparkContext.hadoopFile(SparkContext.scala:556)
at org.apache.spark.SparkContext.textFile(SparkContext.scala:468)
at $line5.$read$iwC$iwC$iwC$iwC.(:13)
at $line5.$read$iwC$iwC$iwC.(:18)
at $line5.$read$iwC$iwC.(:20)
at $line5.$read$iwC.(:22)
at $line5.$read.(:24)
at $line5.$read$.(:28)
at $line5.$read$.()
at $line5.$eval$.(:7)
at $line5.$eval$.()
at $line5.$eval.$print()
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:483)
at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:788)
at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1056)
at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:614)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:645)
at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:609)
at org.apache.spark.repl.SparkILoop.reallyInterpret$1(SparkILoop.scala:796)
at org.apache.spark.repl.SparkILoop.interpretStartingWith(SparkILoop.scala:841)
at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:753)
at org.apache.spark.repl.SparkILoop.processLine$1(SparkILoop.scala:601)
at org.apache.spark.repl.SparkILoop.innerLoop$1(SparkILoop.scala:608)
at org.apache.spark.repl.SparkILoop.loop(SparkILoop.scala:611)
at org.apache.spark.repl.SparkILoop$anonfun$process$1.apply$mcZ$sp(SparkILoop.scala:936)
at org.apache.spark.repl.SparkILoop$anonfun$process$1.apply(SparkILoop.scala:884)
at org.apache.spark.repl.SparkILoop$anonfun$process$1.apply(SparkILoop.scala:884)
at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:884)
at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:982)
at org.apache.spark.repl.Main$.main(Main.scala:31)
at org.apache.spark.repl.Main.main(Main.scala)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:483)
at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:303)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:55)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
14/07/05 19:53:27 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 32.0 KB, free 294.6 MB)
14/07/05 19:53:27 DEBUG BlockManager: Put block broadcast_0 locally took 78 ms
14/07/05 19:53:27 DEBUG BlockManager: Putting block broadcast_0 without replication took 79 ms
res0: org.apache.spark.rdd.RDD[String] = README.md MappedRDD[1] at textFile at :13 复制代码 git同步 对代码作了修改之后，如果并不想提交代码，那该如何将最新的内容同步到本地呢？
git reset --hard
git pull origin master 复制代码 Akka消息跟踪 追踪消息的接收者是谁，相对来说比较容易，只要使用好grep就可以了，当然前提是要对actor model有一点点了解。
还是举个实例吧，我们知道CoarseGrainedSchedulerBackend会发送LaunchTask消息出来，那么谁是接收方呢？只需要执行以下脚本即可。
grep LaunchTask -r core/src/main 复制代码 core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala: case LaunchTask(data) =>
core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala: logError("Received LaunchTask command but executor was null")
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala: case class LaunchTask(data: SerializableBuffer) extends CoarseGrainedClusterMessage
core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala: executorActor(task.executorId) ! LaunchTask(new SerializableBuffer(serializedTask)) 复制代码 小结 今天的内容相对简单，没有技术含量，自己做个记述，免得时间久了，不记得。
相关内容 Apache Spark源码走读之1 -- Spark论文阅读笔记 Apache Spark源码走读之2 -- Job的提交与运行 Apache Spark源码走读之3-- Task运行期之函数调用关系分析 Apache Spark源码走读之4 -- DStream实时流数据处理 Apache Spark源码走读之5-- DStream处理的容错性分析 Apache Spark源码走读之6-- 存储子系统分析 Apache Spark源码走读之7 -- Standalone部署方式分析 Apache Spark源码走读之8 -- Spark on Yarn Apache Spark源码走读之9 -- Spark源码编译 Apache Spark源码走读之10 -- 在YARN上运行SparkPi Apache Spark源码走读之11 -- sql的解析与执行 Apache Spark源码走读之12 -- Hive on Spark运行环境搭建 Apache Spark源码走读之13 -- hiveql on spark实现详解 Apache Spark源码走读之14 -- Graphx实现剖析 Apache Spark源码走读之15 -- Standalone部署模式下的容错性分析 Apache Spark源码走读之16 -- spark repl实现详解 Apache Spark源码走读之18 -- 使用Intellij idea调试Spark源码 Apache Spark源码走读之19 -- standalone cluster模式下资源的申请与释放 Apache Spark源码走读之20 -- ShuffleMapTask计算结果的保存与读取 Apache Spark源码走读之21 -- WEB UI和Metrics初始化及数据更新过程分析 Apache Spark源码走读之22 -- 浅谈mllib中线性回归的算法实现 Apache Spark源码走读之23 -- Spark MLLib中拟牛顿法L-BFGS的源码实现 Apache Spark源码走读之24 -- Sort-based Shuffle的设计与实现
/2 
京ICP备2020039040号
简书网举报电话：021-34700000
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡
