title: Submitting Spark application via YARN client
body: <p>I am using the <code>org.apache.spark.deploy.yarn.Client</code> (Spark 2.1.0) to submit spark a yarn application (SparkPi example). Following are the pertinent lines:</p>
  
  <pre><code>    List&lt;String&gt; arguments = Lists.newArrayList("--class", "org.apache.spark.examples.SparkPi","--jar", "path/to/spark examples jar", "--arg", "10");
  
      SparkConf sparkConf = new SparkConf();
      applicationTag = "TestApp-" + new Date().getTime();
      sparkConf.set("spark.yarn.submit.waitAppCompletion", "false");
      sparkConf.set("spark.yarn.tags", applicationTag);
      sparkConf.set("spark.submit.deployMode", "cluster");
      sparkConf.set("spark.yarn.jars", "/opt/spark/jars/*.jar");
  
      System.setProperty("SPARK_YARN_MODE", "true");
      System.setProperty("SPARK_HOME", "/opt/spark");
  
      ClientArguments cArgs = new ClientArguments(arguments.toArray(new String[arguments.size()]));
      Client client = new Client(cArgs, sparkConf);
      client.run();
  </code></pre>
  
  <p>This seems to be working and the Spark application appears in the YARN RM UI &amp; succeeds. However, the container logs show that the URL for the staging directory is being picked up as 
  <code>SPARK_YARN_STAGING_DIR -&gt; file:/home/{current user}/.sparkStaging/application_xxxxxx</code>. Going through <code>org.apache.spark.deploy.yarn.Client</code> shows the likely reason for it being that the base path for the staging directory is not picked up correctly. The base path should be <code>hdfs://localhost:9000/user/{current user}/</code> rather than <code>file:/home/{current user}/</code> as confirmed by the following error appearing in the logs when the staging directory is cleared off:</p>
  
  <pre><code>java.lang.IllegalArgumentException: Wrong FS: file:/home/user/.sparkStaging/application_1496908076154_0022, expected: hdfs://127.0.0.1:9000
  at org.apache.hadoop.fs.FileSystem.checkPath(FileSystem.java:649)
  at org.apache.hadoop.hdfs.DistributedFileSystem.getPathName(DistributedFileSystem.java:194)
  at org.apache.hadoop.hdfs.DistributedFileSystem.access$000(DistributedFileSystem.java:106)
  at org.apache.hadoop.hdfs.DistributedFileSystem$14.doCall(DistributedFileSystem.java:707)
  at org.apache.hadoop.hdfs.DistributedFileSystem$14.doCall(DistributedFileSystem.java:703)
  at org.apache.hadoop.fs.FileSystemLinkResolver.resolve(FileSystemLinkResolver.java:81)
  at org.apache.hadoop.hdfs.DistributedFileSystem.delete(DistributedFileSystem.java:714)
  at org.apache.spark.deploy.yarn.ApplicationMaster.org$apache$spark$deploy$yarn$ApplicationMaster$$cleanupStagingDir(ApplicationMaster.scala:545)
  at org.apache.spark.deploy.yarn.ApplicationMaster$$anonfun$run$1.apply$mcV$sp(ApplicationMaster.scala:233)
  at org.apache.spark.util.SparkShutdownHook.run(ShutdownHookManager.scala:216)
  at org.apache.spark.util.SparkShutdownHookManager$$anonfun$runAll$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(ShutdownHookManager.scala:188)
  at org.apache.spark.util.SparkShutdownHookManager$$anonfun$runAll$1$$anonfun$apply$mcV$sp$1.apply(ShutdownHookManager.scala:188)
  at org.apache.spark.util.SparkShutdownHookManager$$anonfun$runAll$1$$anonfun$apply$mcV$sp$1.apply(ShutdownHookManager.scala:188)
  at org.apache.spark.util.Utils$.logUncaughtExceptions(Utils.scala:1951)
  at org.apache.spark.util.SparkShutdownHookManager$$anonfun$runAll$1.apply$mcV$sp(ShutdownHookManager.scala:188)
  at org.apache.spark.util.SparkShutdownHookManager$$anonfun$runAll$1.apply(ShutdownHookManager.scala:188)
  at org.apache.spark.util.SparkShutdownHookManager$$anonfun$runAll$1.apply(ShutdownHookManager.scala:188)
  at scala.util.Try$.apply(Try.scala:192)
  at org.apache.spark.util.SparkShutdownHookManager.runAll(ShutdownHookManager.scala:188)
  at org.apache.spark.util.SparkShutdownHookManager$$anon$2.run(ShutdownHookManager.scala:178)
  at org.apache.hadoop.util.ShutdownHookManager$1.run(ShutdownHookManager.java:54)
  </code></pre>
  
  <p>This all works fine when spark-submit is used as I believe it sets all the required environment variables correctly.</p>
  
  <p>I have also tried with setting <code>sparkConf.set("spark.yarn.stagingDir", "hdfs://localhost:9000/user/{current user}");</code> but to no avail as it results in some other errors, such as hdfs not being recognised as a valid file system.</p>