apache-spark - GoogleHadoopFileSystem 无法转换为 hadoop 文件系统？

Question

最初的问题是 trying to deploy spark 1.4 on Google Cloud .下载并设置后

SPARK_HADOOP2_TARBALL_URI='gs://my_bucket/my-images/spark-1.4.1-bin-hadoop2.6.tgz'

使用 bdutil 部署很好；但是，当尝试调用 SqlContext.parquetFile("gs://my_bucket/some_data.parquet") 时，它会遇到以下异常:

 java.lang.ClassCastException: com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem cannot be cast to org.apache.hadoop.fs.FileSystem
at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2595)
at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:91)
at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2630)
at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2612)
at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:370)
at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:169)
at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:354)
at org.apache.hadoop.fs.Path.getFileSystem(Path.java:296)
at org.apache.hadoop.hive.metastore.Warehouse.getFs(Warehouse.java:112)
at org.apache.hadoop.hive.metastore.Warehouse.getDnsPath(Warehouse.java:144)
at org.apache.hadoop.hive.metastore.Warehouse.getWhRoot(Warehouse.java:159)

让我困惑的是，GoogleHadoopFileSystem 应该是 org.apache.hadoop.fs.FileSystem 的子类，我什至在同一个 spark-shell 实例中进行了验证:

scala> var gfs = new com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem()
gfs: com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem = com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem@46f105c

scala> gfs.isInstanceOf[org.apache.hadoop.fs.FileSystem]
res3: Boolean = true

scala> gfs.asInstanceOf[org.apache.hadoop.fs.FileSystem]
res4: org.apache.hadoop.fs.FileSystem = com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem@46f105c

我是否遗漏了什么，有什么解决方法吗？提前致谢!

更新:这是我的 bdutil(版本 1.3.1)部署设置:

import_env hadoop2_env.sh
import_env extensions/spark/spark_env.sh
CONFIGBUCKET="my_conf_bucket"
PROJECT="my_proj"
GCE_IMAGE='debian-7-backports'
GCE_MACHINE_TYPE='n1-highmem-4'
GCE_ZONE='us-central1-f'
GCE_NETWORK='my-network'
GCE_MASTER_MACHINE_TYPE='n1-standard-2'
PREEMPTIBLE_FRACTION=1.0
PREFIX='my-hadoop'
NUM_WORKERS=8
USE_ATTACHED_PDS=true
WORKER_ATTACHED_PDS_SIZE_GB=200
MASTER_ATTACHED_PD_SIZE_GB=200
HADOOP_TARBALL_URI="gs://hadoop-dist/hadoop-2.6.0.tar.gz"
SPARK_MODE="yarn-client"
SPARK_HADOOP2_TARBALL_URI='gs://my_conf_bucket/my-images/spark-1.4.1-bin-hadoop2.6.tgz'

Answer 1

简答题

确实它与 IsolatedClientLoader 有关，我们已经追查了根本原因并验证了修复。我提交了 https://issues.apache.org/jira/browse/SPARK-9206跟踪这个问题，并通过一个简单的修复成功地从我的 fork 构建了一个干净的 Spark tarball:https://github.com/apache/spark/pull/7549

有一些短期选择:

1. 现在使用 Spark 1.3.1。
2. 在您的 bdutil 部署中，使用 HDFS 作为默认文件系统 (--default_fs=hdfs)；您仍然可以在作业中直接指定 gs:// 路径，只是 HDFS 将用于中间数据和暂存文件。不过，在此模式下使用原始 Hive 存在一些小的不兼容性。
3. 如果不需要 HiveContext 功能，请使用原始 val sqlContext = new org.apache.spark.sql.SQLContext(sc) 而不是 HiveContext。
4. git clone https://github.com/dennishuo/spark 并运行 ./make-distribution.sh --name my-custom-spark --tgz --skip- java-test -Pyarn -Phadoop-2.6 -Dhadoop.version=2.6.0 -Phive -Phive-thriftserver 获取新的压缩包，您可以在 bdutil 的 spark_env.sh 中指定。

长答案

我们已经验证它仅在 fs.default.name 和 fs.defaultFS 设置为 gs:// 时出现路径，无论是否尝试从 parquetFile("gs://...") 或 parquetFile("hdfs://...") 加载路径，当 fs.default.name 和 fs.defaultFS 设置为 HDFS 路径时，从 HDFS 和 GCS 加载数据工作正常。这也是目前 Spark 1.4+ 特有的，在 Spark 1.3.1 或更早版本中不存在。

回归似乎已在 https://github.com/apache/spark/commit/9ac8393663d759860c67799e000ec072ced76493 中引入这实际上修复了先前相关的类加载问题，SPARK-8368 .虽然修复本身对于正常情况是正确的，但有一种方法 IsolatedClientLoader.isSharedClass用于确定使用哪个类加载器，并与上述提交交互以破坏 GoogleHadoopFileSystem 类加载。

该文件中的以下行包括 com.google.* 下的所有内容作为“共享类”，因为 Guava 和可能的 protobuf 依赖项确实作为共享库加载，但不幸的是 GoogleHadoopFileSystem 应该是在这种情况下作为“配置单元类”加载，就像 org.apache.hadoop.hdfs.DistributedFileSystem 一样。我们只是碰巧共享了 com.google.* 包命名空间。

protected def isSharedClass(name: String): Boolean =
  name.contains("slf4j") ||
  name.contains("log4j") ||
  name.startsWith("org.apache.spark.") ||
  name.startsWith("scala.") ||
  name.startsWith("com.google") ||
  name.startsWith("java.lang.") ||
  name.startsWith("java.net") ||
  sharedPrefixes.exists(name.startsWith)

...

/** The classloader that is used to load an isolated version of Hive. */
protected val classLoader: ClassLoader = new URLClassLoader(allJars, rootClassLoader) {
  override def loadClass(name: String, resolve: Boolean): Class[_] = {
    val loaded = findLoadedClass(name)
    if (loaded == null) doLoadClass(name, resolve) else loaded
  }

  def doLoadClass(name: String, resolve: Boolean): Class[_] = {
    ...
    } else if (!isSharedClass(name)) {
      logDebug(s"hive class: $name - ${getResource(classToPath(name))}")
      super.loadClass(name, resolve)
    } else {
      // For shared classes, we delegate to baseClassLoader.
      logDebug(s"shared class: $name")
      baseClassLoader.loadClass(name)
    }
  }
}

这可以通过将以下行添加到 ${SPARK_INSTALL}/conf/log4j.properties 来验证:

log4j.logger.org.apache.spark.sql.hive.client=DEBUG

输出显示:

...
15/07/20 20:59:14 DEBUG IsolatedClientLoader: hive class: org.apache.hadoop.hdfs.DistributedFileSystem - jar:file:/home/hadoop/spark-install/lib/spark-assembly-1.4.1-hadoop2.6.0.jar!/org/apache/hadoop/hdfs/DistributedFileSystem.class
...
15/07/20 20:59:14 DEBUG IsolatedClientLoader: shared class: com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem
java.lang.RuntimeException: java.lang.ClassCastException: com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem cannot be cast to org.apache.hadoop.fs.FileSystem