{"action":"created","issue":{"url":"https://api.github.com/repos/spark-jobserver/spark-jobserver/issues/12","labels_url":"https://api.github.com/repos/spark-jobserver/spark-jobserver/issues/12/labels{/name}","comments_url":"https://api.github.com/repos/spark-jobserver/spark-jobserver/issues/12/comments","events_url":"https://api.github.com/repos/spark-jobserver/spark-jobserver/issues/12/events","html_url":"https://github.com/spark-jobserver/spark-jobserver/issues/12","id":43504758,"number":12,"title":"Fail to read HDFS text file then saveAsHdfsFile from spark-jobserver job due to HDFS _temporary folder not found. ","user":{"login":"svendx4f","id":1214071,"avatar_url":"https://avatars.githubusercontent.com/u/1214071?v=3","gravatar_id":"","url":"https://api.github.com/users/svendx4f","html_url":"https://github.com/svendx4f","followers_url":"https://api.github.com/users/svendx4f/followers","following_url":"https://api.github.com/users/svendx4f/following{/other_user}","gists_url":"https://api.github.com/users/svendx4f/gists{/gist_id}","starred_url":"https://api.github.com/users/svendx4f/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/svendx4f/subscriptions","organizations_url":"https://api.github.com/users/svendx4f/orgs","repos_url":"https://api.github.com/users/svendx4f/repos","events_url":"https://api.github.com/users/svendx4f/events{/privacy}","received_events_url":"https://api.github.com/users/svendx4f/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":4,"created_at":"2014-09-22T15:42:12Z","updated_at":"2015-10-07T06:09:18Z","closed_at":"2015-10-07T06:09:18Z","body":"I am failing to execute a basic read from HDFS / process / write to HDFS Spark job executed from the Spark job server. \r\n\r\nWe are running the latest version of the job-server from the master branch with Spark 1.1, Mesos 0.20 and HDFS from CDH 5.1 (the same issue is visible with Spark 1.0.2). \r\n\r\nThe code of the process(..) method works successfully on Spark/Mesos when executed from the spark-shell but fails when bundled in a SparkJob and ran from the Spark job server. \r\n\r\n\r\n```\r\nclass JobServerTest extends SparkJob {\r\n\r\n  override def runJob(sparkContext: SparkContext, jobConfig: Config): Any =\r\n    new Tool().process(sparkContext, jobConfig.getString(\"destinationFoler\"))\r\n\r\n  override def validate(sc: SparkContext, config: Config): SparkJobValidation = SparkJobValid\r\n\r\n}\r\n\r\nclass Tool extends Serializable{\r\n\r\n  case class Data (part: Int, word1: String, amount1: Double, amount2: Double, letter1: String, word2: String) extends Serializable\r\n\r\n  def process(sparkContext: SparkContext, destinationFolder: String) =\r\n    sparkContext.textFile(\"hdfs://vm28-hulk-priv:8020/svend/test-jobserver/source\")\r\n      .map(_.split(\",\"))\r\n      .filter(_.length == 6)\r\n      .map(raw => new Data(raw(0).toInt, raw(1), raw(2).toDouble, raw(3).toDouble, raw(4), raw(5)) )\r\n\r\n      // some silly operations just to shuffle + process data\r\n      .map (data => data.copy(amount1 = data.amount2 * Math.sqrt(data.amount1)))\r\n      .groupBy(_.part)\r\n      .flatMap { case (part, data) => data}\r\n      .saveAsTextFile(destinationFolder)\r\n}\r\n```\r\n\r\n\r\nOn each Mesos slave nodes, the following stack traces are visible: \r\n\r\n```\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Getting 16 non-empty blocks out of 16 blocks\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 0 remote fetches in 6 ms\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 0 remote fetches in 2 ms\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 0 remote fetches in 1 ms\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 2 remote fetches in 15 ms\r\n14/09/22 11:27:33 INFO SendingConnection: Initiating connection to [vm21-hulk-priv.mtl.mnubo.com/10.237.241.21:54104]\r\n14/09/22 11:27:33 INFO SendingConnection: Connected to [vm21-hulk-priv.mtl.mnubo.com/10.237.241.21:54104], 2 messages pending\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 2 remote fetches in 12 ms\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 2 remote fetches in 16 ms\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 2 remote fetches in 16 ms\r\n14/09/22 11:27:33 INFO ConnectionManager: Accepted connection from [vm21-hulk-priv.mtl.mnubo.com/10.237.241.21:32981]\r\n14/09/22 11:27:33 INFO BlockFetcherIterator$BasicBlockFetcherIterator: Started 2 remote fetches in 22 ms\r\n14/09/22 11:27:33 ERROR Executor: Exception in task 12.0 in stage 0.0 (TID 28)\r\njava.io.IOException: The temporary job-output directory file:/svend/test-jobserver/dest6/_temporary doesn't exist!\r\n\tat org.apache.hadoop.mapred.FileOutputCommitter.getWorkPath(FileOutputCommitter.java:250)\r\n\tat org.apache.hadoop.mapred.FileOutputFormat.getTaskOutputPath(FileOutputFormat.java:240)\r\n\tat org.apache.hadoop.mapred.TextOutputFormat.getRecordWriter(TextOutputFormat.java:116)\r\n\tat org.apache.spark.SparkHadoopWriter.open(SparkHadoopWriter.scala:89)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:980)\r\n\tat org.apache.spark.rdd.PairRDDFunctions$$anonfun$13.apply(PairRDDFunctions.scala:974)\r\n\tat org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:62)\r\n\tat org.apache.spark.scheduler.Task.run(Task.scala:54)\r\n\tat org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:177)\r\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)\r\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)\r\n\tat java.lang.Thread.run(Thread.java:744)\r\n14/09/22 11:27:33 ERROR Executor: Exception in tas\r\n```\r\n\r\nThe data read in that folder is just a bunch of random csv lines, generated as follows (from spark-shell): \r\n\r\n```\r\nclass generator() {\r\n\r\n    val letters = 'a' to 'z'        \r\n    val rand = new scala.util.Random()\r\n\r\n    def letter = letters(rand.nextInt(26))\r\n    def word = 1 to rand.nextInt(15) map (_ => letter) mkString\r\n    def amount = rand.nextGaussian * rand.nextInt(100)\r\n    def part = rand.nextInt(10)\r\n\r\n    def line = part :: word :: amount :: amount :: letter :: word :: Nil mkString(\",\")\r\n    def lines = (1 to 500000).foldLeft(List[String]()) { case (acc, _) => line :: acc } \r\n\r\n}\r\n\r\nsc.parallelize(1 to 16, 16).map (_ => new generator()).flatMap (_.lines).saveAsTextFile(\"/svend/test-jobserver/source\")\r\n\r\n```\r\n\r\n\r\n\r\n\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/spark-jobserver/spark-jobserver/issues/comments/146087305","html_url":"https://github.com/spark-jobserver/spark-jobserver/issues/12#issuecomment-146087305","issue_url":"https://api.github.com/repos/spark-jobserver/spark-jobserver/issues/12","id":146087305,"user":{"login":"velvia","id":1062875,"avatar_url":"https://avatars.githubusercontent.com/u/1062875?v=3","gravatar_id":"","url":"https://api.github.com/users/velvia","html_url":"https://github.com/velvia","followers_url":"https://api.github.com/users/velvia/followers","following_url":"https://api.github.com/users/velvia/following{/other_user}","gists_url":"https://api.github.com/users/velvia/gists{/gist_id}","starred_url":"https://api.github.com/users/velvia/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/velvia/subscriptions","organizations_url":"https://api.github.com/users/velvia/orgs","repos_url":"https://api.github.com/users/velvia/repos","events_url":"https://api.github.com/users/velvia/events{/privacy}","received_events_url":"https://api.github.com/users/velvia/received_events","type":"User","site_admin":false},"created_at":"2015-10-07T06:09:18Z","updated_at":"2015-10-07T06:09:18Z","body":"Closing, feel free to reopen if this still happens with latest Spark and master."}}