許鵬：以Spark為例淺談源碼跟讀實踐

【編者按】在 對許鵬的采訪中，我們有從方法上進行了大型開源項目的學習，其中包括Problem domain→model→architecture&implementation→improvement→best practice的思維范式，而本次許鵬的博文則更關注源碼跟讀過程中消息或調用的流程追蹤。

免費訂閱“CSDN云計算”微信公眾號，實時掌握第一手云中消息！

CSDN作為國內最專業的云計算服務平臺，提供云計算、大數據、虛擬化、數據中心、OpenStack、CloudStack、Hadoop、Spark、機器學習、智能算法等相關云計算觀點，云計算技術，云計算平臺，云計算實踐，云計算產業資訊等服務。

下為原文

概要

本次不談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()

然后，重新進行源碼編譯

sbt/sbt assembly

再次打開spark-shell,執行sc.textFile("README.md"),就可以得到如下輸出，從中可以清楚知道tryToPut的調用者是誰

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

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源碼走讀之17 -- 如何進行代碼跟讀（責編/仲浩）

生活不易，碼農辛苦
如果您覺得本網站對您的學習有所幫助,可以手機掃描二維碼進行捐贈