Spark SQL on Yarn with Cluster m

问题描述

运行spark sql on yarn的时候发现yarn client模式跑的好好的程序，换成yarn cluster模式就不正确了，原因是hive-site.xml这文件没有被加载到Driver（也就是这时候的ApplicationMaster）的classpath里面去，貌似是直接连接了一个默认的am-container本地metastore。

看下官方文档 2.1.2 - 2.1.1 - 2.0.2 貌似都是同样一句话将hive-site.xml放入conf/下就行了。我也真真的造作了啊。。

Configuration of Hive is done by placing your hive-site.xml, core-site.xml (for security configuration), and hdfs-site.xml (for HDFS configuration) file in conf/.

测试程序

如下，简单到羞涩

import org.apache.spark.sql.SparkSession
object ShowHiveTables {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("Show Hive Tables")
      .enableHiveSupport()
      .getOrCreate()
    spark.sql("show tables").show()
    spark.stop()
  }
}

扒下spark源码

$SPARK_HOME/conf下的.xml文件果然是不会上传的 基于这个commit: b04eefae49b96e2ef5a8d75334db29ef4e19ce58给出
org.apache.spark.deploy.yarn.Client

/**
   * Create an archive with the config files for distribution.
   *
   * These will be used by AM and executors. The files are zipped and added to the job as an
   * archive, so that YARN will explode it when distributing to AM and executors. This directory
   * is then added to the classpath of AM and executor process, just to make sure that everybody
   * is using the same default config.
   *
   * This follows the order of precedence set by the startup scripts, in which HADOOP_CONF_DIR
   * shows up in the classpath before YARN_CONF_DIR.
   *
   * Currently this makes a shallow copy of the conf directory. If there are cases where a
   * Hadoop config directory contains subdirectories, this code will have to be fixed.
   *
   * The archive also contains some Spark configuration. Namely, it saves the contents of
   * SparkConf in a file to be loaded by the AM process.
   */
  private def createConfArchive(): File = {
    
    val hadoopConfFiles = new HashMap[String, File]()
    // 处理了HADOOP_CONF_DIR下的配置文件
    Seq("HADOOP_CONF_DIR", "YARN_CONF_DIR").foreach { envKey =>
      sys.env.get(envKey).foreach { path =>
        val dir = new File(path)
        if (dir.isDirectory()) {
          val files = dir.listFiles()
          if (files == null) {
            logWarning("Failed to list files under directory " + dir)
          } else {
            files.foreach { file =>
              if (file.isFile && !hadoopConfFiles.contains(file.getName())) {
                hadoopConfFiles(file.getName()) = file
              }
            }
          }
        }
      }
    }

    val confArchive = File.createTempFile(LOCALIZED_CONF_DIR, ".zip",
      new File(Utils.getLocalDir(sparkConf)))
    val confStream = new ZipOutputStream(new FileOutputStream(confArchive))

    try {
      confStream.setLevel(0)

      // Upload $SPARK_CONF_DIR/log4j.properties file to the distributed cache to make sure that
      // the executors will use the latest configurations instead of the default values. This is
      // required when user changes log4j.properties directly to set the log configurations. If
      // configuration file is provided through --files then executors will be taking configurations
      // from --files instead of $SPARK_CONF_DIR/log4j.properties.

      // Also upload metrics.properties to distributed cache if exists in classpath.
      // If user specify this file using --files then executors will use the one
      // from --files instead.
      //  处理$SPARK_CONF_DIR下的文件，但是只有下面这两个，没有hive-site.xml
      for { prop <- Seq("log4j.properties", "metrics.properties")
            url <- Option(Utils.getContextOrSparkClassLoader.getResource(prop))
            if url.getProtocol == "file" } {
        val file = new File(url.getPath())
        confStream.putNextEntry(new ZipEntry(file.getName()))
        Files.copy(file, confStream)
        confStream.closeEntry()
      }

      // Save the Hadoop config files under a separate directory in the archive. This directory
      // is appended to the classpath so that the cluster-provided configuration takes precedence.
      confStream.putNextEntry(new ZipEntry(s"$LOCALIZED_HADOOP_CONF_DIR/"))
      confStream.closeEntry()
      hadoopConfFiles.foreach { case (name, file) =>
        if (file.canRead()) {
          confStream.putNextEntry(new ZipEntry(s"$LOCALIZED_HADOOP_CONF_DIR/$name"))
          Files.copy(file, confStream)
          confStream.closeEntry()
        }
      }

      // Save the YARN configuration into a separate file that will be overlayed on top of the
      // cluster's Hadoop conf.
      confStream.putNextEntry(new ZipEntry(SPARK_HADOOP_CONF_FILE))
      yarnConf.writeXml(confStream)
      confStream.closeEntry()

      // Save Spark configuration to a file in the archive.
      val props = new Properties()
      sparkConf.getAll.foreach { case (k, v) => props.setProperty(k, v) }
      // Override spark.yarn.key to point to the location in distributed cache which will be used
      // by AM.
      Option(amKeytabFileName).foreach { k => props.setProperty(KEYTAB.key, k) }
      confStream.putNextEntry(new ZipEntry(SPARK_CONF_FILE))
      val writer = new OutputStreamWriter(confStream, StandardCharsets.UTF_8)
      props.store(writer, "Spark configuration.")
      writer.flush()
      confStream.closeEntry()
    } finally {
      confStream.close()
    }
    confArchive
  }

目测是个bug

顺手提交个代码

https://github.com/apache/spark/pull/19663

继续解决问题

1. --files path/to/your/hive-site.xml 放到了am container的工作目录，有效
2. --jars path/to/your/hive-site.xml 当成jar包上传，有效
3. cp path/to/your/hive-site.xml $HADOOP_CONF_DIR，看上面代码，有效
4. --conf spark.yarn.dist.files=path/to/your/hive-site.xml 同1，有效
5. --conf spark.yarn.dist.jars=path/to/your/hive-site.xml 同2，有效

测试版本在2.1.x下进行，其他版本自行验证，或者直接打上上面的patch