GoogleHadoopFileSystem не может быть приведен к hadoop FileSystem?

Первоначальный вопрос заключался в попытке развернуть спарк 1.4 в 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'
Источник

1 ответ

Решение

Короткий ответ

Действительно, это было связано с IsolatedClientLoader, и мы отследили основную причину и проверили исправление. Я подал https://issues.apache.org/jira/browse/SPARK-9206 чтобы отследить эту проблему, и успешно построил чистый тарбол Spark из моей вилки с простым исправлением: https://github.com/apache/spark/pull/7549

Есть несколько краткосрочных вариантов:

  1. Используйте Spark 1.3.1 сейчас.
  2. В вашем развертывании bdutil используйте HDFS в качестве файловой системы по умолчанию (--default_fs=hdfs); вы все равно сможете напрямую указать gs:// пути в ваших заданиях, просто HDFS будет использоваться для промежуточных данных и промежуточных файлов. Однако в этом режиме есть некоторые незначительные несовместимости с использованием необработанного Hive.
  3. Используйте сырье val sqlContext = new org.apache.spark.sql.SQLContext(sc) вместо HiveContext, если вам не нужны функции 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's 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
Источник
Другие вопросы по тегам