2

Mysteriously, when I use Spark my custom filesystem provider vanishes.

The full source for my example is available on github so you can follow.

I'm using maven to depend on gcloud-java-nio, which provides a Java FileSystem for Google Cloud Storage, via "gs://" URLs. My Spark project uses maven-shade-plugin to create one big jar with all the source in it.

The big jar correctly includes a META-INF/services/java.nio.file.spi.FileSystemProvider file, containing the correct name for the class (com.google.cloud.storage.contrib.nio.CloudStorageFileSystemProvider). I checked and that class is also correctly included in the jar file.

The program uses FileSystemProvider.installedProviders() to list the filesystem providers it finds. "gs" should be listed (and it is if I run the same function in a non-Spark context), but when running with Spark on Dataproc, that provider's gone.

I'd like to know: How can I use a custom filesystem in my Spark program?

edit: Dennis Huo helpfully contributed that he sees the same problem when running on a Spark cluster, so the problem isn't specific to Dataproc. In fact, it also occurs when just using Scala. Also there are workarounds for the example I'm showing here, but I'd still like to know how to use a custom filesystem with Spark.

2 Answers 2

4

This doesn't appear to be a Dataproc-specific issue, more of a Scala issue and Spark fundamentally depends on Scala; if I build your jarfile and then load it up with scala independently of Dataproc or Spark, I get:

scala -cp spark-repro-1.0-SNAPSHOT.jar

scala> import java.nio.file.spi.FileSystemProvider
import java.nio.file.spi.FileSystemProvider

scala> import scala.collection.JavaConversions._
import scala.collection.JavaConversions._

scala> FileSystemProvider.installedProviders().toList.foreach(l => println(l.getScheme()))
file
jar

scala> SparkRepro.listFS(1)
res3: String = Worker 1 installed filesystem providers: file jar

So it seems any bundling that's being done isn't properly registering the FileSystem provider, at least for scala. I tested the theory using the ListFilesystems example code (just removed the package at the top for convenience) on both a Dataproc node as well as a manually created VM with scala and Java 7 installed independently of Dataproc just to double-check.

$ cat ListFilesystems.java
import java.io.IOException;
import java.net.URI;
import java.nio.file.FileSystem;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.spi.FileSystemProvider;

/**
 * ListFilesystems is a super-simple program that lists the available NIO filesystems.
 */
public class ListFilesystems {

  /**
   * See the class documentation.
   */
  public static void main(String[] args) throws IOException {
    listFilesystems();
  }

  private static void listFilesystems() {
    System.out.println("Installed filesystem providers:");
    for (FileSystemProvider p : FileSystemProvider.installedProviders()) {
      System.out.println("  " + p.getScheme());
    }
  }

}

$ javac ListFilesystems.java

Running using java and then scala:

$ java -cp spark-repro-1.0-SNAPSHOT.jar:. ListFilesystems
Installed filesystem providers:
  file
  jar
  gs
$ scala -cp spark-repro-1.0-SNAPSHOT.jar:. ListFilesystems
Installed filesystem providers:
  file
  jar
$

This was the same on both Dataproc and my non-Dataproc VM. It looks like there's still unresolved difficulties getting the FileSystemProviders to load properly in Scala, and there doesn't seem to be an easy way to dynamically register them system-wide at runtime either; the most I could find was this old thread that didn't seem to come to any useful conclusion.

Fortunately though, it looks like at least the CloudStorageFileSystemProvider has no problem making it onto the classpath, so you can at least fall back to explicitly creating an instance of the cloud storage provider to use:

new com.google.cloud.storage.contrib.nio.CloudStorageFileSystemProvider()
    .getFileSystem(new java.net.URI("gs://my-bucket"))

Alternatively, if you're using Spark anyways, you might want to consider just using the Hadoop FileSystem interfaces. It's very similar to Java NIO FileSystems (rather, predated the Java NIO FileSystem stuff), and it's more portable for now. You can easily do things like:

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.conf.Configuration;

...
Path foo = new Path("gs://my-bucket/my-data.txt");
InputStream is = foo.getFileSystem(new Configuration()).open(foo);
...

The benefit of working with the Hadoop FileSystem interfaces is that you're guaranteed the configuration/settings will be clean both in your driver program and in the distributed worker nodes. For example, sometimes you'll need to modify filesystem settings just for a single job running in a Dataproc cluster; then you can plumb through Hadoop properties which are properly scoped for a single job without interfering with other jobs running at the same time.

3
  • Thanks for the detailed answer and workaround. I would still like to figure out how to load a custom filesystem onto Spark, though! Commented Sep 15, 2016 at 16:41
  • You may also find more success expanding the tags to pose this as a general Scala question, since I suspect if you can get "scala -cp" to do the right thing, you'll be able to get Spark to do so as well. This question is seemingly related but as you mention, the jarfile definitely contains the right META-INF/services file already, so the problem is different.
    – Dennis Huo
    Commented Sep 15, 2016 at 17:48
  • I don't know about your setup, but there's no registration: you just need to put your jar on the system class path to be visible to the system class loader. It might also depend on your version of spark. See my answer for the easy case.
    – som-snytt
    Commented Apr 7, 2017 at 18:23
0

My comment on the linked ticket (https://github.com/scala/bug/issues/10247):

Using scala -toolcp path makes your app jar available to the system class loader.

Or, use the API where you can provide the class loader to find providers which are not "installed."

The scala runner script has a few interacting parts, like -Dscala.usejavacp and -nobootcp, which possibly behave differently on Windows. It's not always obvious which incantation to use.

The misunderstanding here is the assumption that java and scala do the same thing with respect to -cp.

This example shows loading a test provider from a build dir. The "default" provider comes first, hence the strange ordering.

$ skala -toolcp ~/bin
Welcome to Scala 2.12.2 (OpenJDK 64-Bit Server VM 1.8.0_112)

scala> import java.nio.file.spi.FileSystemProvider
import java.nio.file.spi.FileSystemProvider

scala> FileSystemProvider.installedProviders
res0: java.util.List[java.nio.file.spi.FileSystemProvider] = [sun.nio.fs.LinuxFileSystemProvider@12abdfb, com.acme.FlakeyFileSystemProvider@b0e5507, com.acme.FlakeyTPDFileSystemProvider@6bbe50c9, com.sun.nio.zipfs.ZipFileSystemProvider@3c46dcbe]

scala> :quit

Or specifying loader:

$ skala -cp ~/bin
Welcome to Scala 2.12.2 (OpenJDK 64-Bit Server VM 1.8.0_112)

scala> import java.net.URI
import java.net.URI

scala> val uri = URI.create("tpd:///?count=10000")
uri: java.net.URI = tpd:///?count=10000

scala> import collection.JavaConverters._
import collection.JavaConverters._

scala> val em = Map.empty[String, AnyRef].asJava
em: java.util.Map[String,AnyRef] = {}

scala> import java.nio.file.FileSystems
import java.nio.file.FileSystems

scala> FileSystems.
getDefault   getFileSystem   newFileSystem

scala> FileSystems.newFileSystem(uri, em, $intp.classLoader)
res1: java.nio.file.FileSystem = com.acme.FlakeyFileSystemProvider$FlakeyFileSystem@2553dcc0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.