diff options
author | Karen Feng <karenfeng.us@gmail.com> | 2013-07-10 10:20:53 -0700 |
---|---|---|
committer | Karen Feng <karenfeng.us@gmail.com> | 2013-07-10 10:20:53 -0700 |
commit | 620a6974c6603f1c0e5a7cea8f0387a5d18f2e5e (patch) | |
tree | 35023419331ca3cf7477ccb4a8e5d37023b3a766 /core/src/main | |
parent | b6072b58bf795093df0e5f8424413fcd0cce6323 (diff) | |
download | spark-620a6974c6603f1c0e5a7cea8f0387a5d18f2e5e.tar.gz spark-620a6974c6603f1c0e5a7cea8f0387a5d18f2e5e.tar.bz2 spark-620a6974c6603f1c0e5a7cea8f0387a5d18f2e5e.zip |
Allows for larger files, refactors lastNBytes, removes old Log column, fixes imports, uses map
Diffstat (limited to 'core/src/main')
4 files changed, 15 insertions, 37 deletions
diff --git a/core/src/main/scala/spark/Utils.scala b/core/src/main/scala/spark/Utils.scala index dfc30469cf..849edc13ee 100644 --- a/core/src/main/scala/spark/Utils.scala +++ b/core/src/main/scala/spark/Utils.scala @@ -621,12 +621,12 @@ private object Utils extends Logging { callSiteInfo.firstUserLine) } - /** Return a string containing the last `n` bytes of a file. */ - def lastNBytes(path: String, n: Int): String = { + /** Return a string containing part of a file from byte 'a' to 'b'. */ + def offsetBytes(path: String, a: Long, b: Long): String = { val file = new File(path) val length = file.length() - val buff = new Array[Byte](math.min(n, length.toInt)) - val skip = math.max(0, length - n) + val buff = new Array[Byte](math.min((b-a).toInt, length.toInt)) + val skip = math.max(0, a) val stream = new FileInputStream(file) stream.skip(skip) @@ -635,18 +635,11 @@ private object Utils extends Logging { Source.fromBytes(buff).mkString } - /** Return an array containing part of a file from byte 'a' to 'b'. */ - def offsetBytes(path: String, a: Int, b: Int): String = { + /** Return a string containing the last `n` bytes of a file. */ + def lastNBytes(path: String, n: Int): String = { val file = new File(path) val length = file.length() - val buff = new Array[Byte](math.min(b-a, length.toInt)) - val skip = math.max(0, a) - val stream = new FileInputStream(file) - - stream.skip(skip) - stream.read(buff) - stream.close() - Source.fromBytes(buff).mkString + offsetBytes(path, length-n, length) } /** diff --git a/core/src/main/scala/spark/deploy/master/ui/ApplicationPage.scala b/core/src/main/scala/spark/deploy/master/ui/ApplicationPage.scala index cc32728c1c..49ced0d320 100644 --- a/core/src/main/scala/spark/deploy/master/ui/ApplicationPage.scala +++ b/core/src/main/scala/spark/deploy/master/ui/ApplicationPage.scala @@ -38,7 +38,7 @@ private[spark] class ApplicationPage(parent: MasterWebUI) { state.completedApps.find(_.id == appId).getOrElse(null) }) - val executorHeaders = Seq("ExecutorID", "Worker", "Cores", "Memory", "State", "Logs", "Log Pages") + val executorHeaders = Seq("ExecutorID", "Worker", "Cores", "Memory", "State", "Log Pages") val executors = app.executors.values.toSeq val executorTable = UIUtils.listingTable(executorHeaders, executorRow, executors) @@ -90,16 +90,10 @@ private[spark] class ApplicationPage(parent: MasterWebUI) { <td>{executor.memory}</td> <td>{executor.state}</td> <td> - <a href={"%s/log?appId=%s&executorId=%s&logType=stdout" - .format(executor.worker.webUiAddress, executor.application.id, executor.id)}>stdout</a> - <a href={"%s/log?appId=%s&executorId=%s&logType=stderr" - .format(executor.worker.webUiAddress, executor.application.id, executor.id)}>stderr</a> - </td> - <td> <a href={"%s/logPage?appId=%s&executorId=%s&logType=stdout&offset=0&byteLength=2000" - .format(executor.worker.webUiAddress, executor.application.id, executor.id)}>stdout-page</a> + .format(executor.worker.webUiAddress, executor.application.id, executor.id)}>stdout</a> <a href={"%s/logPage?appId=%s&executorId=%s&logType=stderr&offset=0&byteLength=2000" - .format(executor.worker.webUiAddress, executor.application.id, executor.id)}>stderr-page</a> + .format(executor.worker.webUiAddress, executor.application.id, executor.id)}>stderr</a> </td> </tr> } diff --git a/core/src/main/scala/spark/deploy/worker/ui/IndexPage.scala b/core/src/main/scala/spark/deploy/worker/ui/IndexPage.scala index d532aa9e95..af9943853f 100644 --- a/core/src/main/scala/spark/deploy/worker/ui/IndexPage.scala +++ b/core/src/main/scala/spark/deploy/worker/ui/IndexPage.scala @@ -29,7 +29,7 @@ private[spark] class IndexPage(parent: WorkerWebUI) { val stateFuture = (worker ? RequestWorkerState)(timeout).mapTo[WorkerState] val workerState = Await.result(stateFuture, 30 seconds) - val executorHeaders = Seq("ExecutorID", "Cores", "Memory", "Job Details", "Logs", "Log Pages") + val executorHeaders = Seq("ExecutorID", "Cores", "Memory", "Job Details", "Log Pages") val runningExecutorTable = UIUtils.listingTable(executorHeaders, executorRow, workerState.executors) val finishedExecutorTable = @@ -88,16 +88,10 @@ private[spark] class IndexPage(parent: WorkerWebUI) { </ul> </td> <td> - <a href={"log?appId=%s&executorId=%s&logType=stdout" - .format(executor.appId, executor.execId)}>stdout</a> - <a href={"log?appId=%s&executorId=%s&logType=stderr" - .format(executor.appId, executor.execId)}>stderr</a> - </td> - <td> <a href={"logPage?appId=%s&executorId=%s&logType=stdout&offset=0&byteLength=2000" - .format(executor.appId, executor.execId)}>stdout-page</a> + .format(executor.appId, executor.execId)}>stdout</a> <a href={"logPage?appId=%s&executorId=%s&logType=stderr&offset=0&byteLength=2000" - .format(executor.appId, executor.execId)}>stderr-page</a> + .format(executor.appId, executor.execId)}>stderr</a> </td> </tr> } diff --git a/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala b/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala index b018f80c85..602881d5f1 100644 --- a/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala +++ b/core/src/main/scala/spark/deploy/worker/ui/WorkerWebUI.scala @@ -9,9 +9,6 @@ import javax.servlet.http.HttpServletRequest import org.eclipse.jetty.server.{Handler, Server} -import scala.io.Source._ -import scala.xml._ - import spark.{Utils, Logging} import spark.ui.JettyUtils import spark.ui.JettyUtils._ @@ -74,11 +71,11 @@ class WorkerWebUI(val worker: ActorRef, val workDir: File, requestedPort: Option val appId = request.getParameter("appId") val executorId = request.getParameter("executorId") val logType = request.getParameter("logType") - val offset = Option(request.getParameter("offset")).map(_.toInt).getOrElse(0) + val offset = Option(request.getParameter("offset")).map(_.toLong).getOrElse(0).asInstanceOf[Long] val maxBytes = 1024 * 1024 val defaultBytes = 100 * 1024 - val byteLength = Option(request.getParameter("byteLength")).flatMap(s => Some(s.toInt)).getOrElse(defaultBytes) + val byteLength = Option(request.getParameter("byteLength")).map(_.toInt).getOrElse(defaultBytes) val path = "%s/%s/%s/%s".format(workDir.getPath, appId, executorId, logType) val logLength = new File(path).length() |