Spark-on-k8s-operator: spec.deps does not support s3 out of the box

[olegfr]

If I put dependency for a file in s3a, for example:

mainApplicationFile: s3a://riskflow-dev/spark-examples/pi.py
deps:
files:
- s3a://riskflow-dev/spark-examples/rfstorage.py

spark-operator fails to start driver pod. Please see the full Exception in the bottom. My image registry.ng.bluemix.net/riskflow/spark-py:14-spark-2.4.0 contains extra jars that allow its spark to access s3a. It looks to me that the reason why this error occurs, is that Spark-Operator image does not have these jars in its spark (/opt/spark/jars/)

I know for sure the issue is not in my image, since if I remove deps.files from the yaml and leave s3a in only mainApplicationFile, then everything works fine.

It looks that spark on the Spark-Operator pod downloads all the remote dependencies. In such a case, how am I supposed to configure/add jars to it in order to access s3?

kubectl logs queenly-starfish-sparkoperator-56766986-t2rfc --namespace spark-operator
I0213 20:21:29.486895 11 submission.go:64] spark-submit arguments: [/opt/spark/bin/spark-submit --master k8s://https://172.21.0.1:443 --deploy-mode cluster --conf spark.kubernetes.namespace=default --conf spark.app.name=pyspark-pi --conf spark.kubernetes.driver.pod.name=pyspark-pi-driver --files s3a://riskflow-dev/spark-examples/rfstorage.py --conf spark.kubernetes.container.image=registry.ng.bluemix.net/riskflow/spark-py:14-spark-2.4.0 --conf spark.kubernetes.container.image.pullPolicy=Always --conf spark.kubernetes.container.image.pullSecrets=bluemix-default-secret --conf spark.kubernetes.pyspark.pythonVersion=2 --conf spark.kubernetes.submission.waitAppCompletion=false --conf spark.hadoop.fs.s3a.secret.key= --conf spark.hadoop.fs.s3a.access.key= --conf spark.hadoop.fs.s3a.endpoint= --conf spark.kubernetes.driver.label.sparkoperator.k8s.io/app-name=pyspark-pi --conf spark.kubernetes.driver.label.sparkoperator.k8s.io/launched-by-spark-operator=true --conf spark.kubernetes.driver.pod.name=pyspark-pi-driver --conf spark.driver.cores=0.100000 --conf spark.kubernetes.driver.limit.cores=200m --conf spark.driver.memory=512m --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark --conf spark.kubernetes.driver.label.version=2.4.0 --conf spark.kubernetes.executor.label.sparkoperator.k8s.io/app-name=pyspark-pi --conf spark.kubernetes.executor.label.sparkoperator.k8s.io/launched-by-spark-operator=true --conf spark.executor.instances=1 --conf spark.executor.cores=1 --conf spark.executor.memory=512m --conf spark.kubernetes.executor.label.version=2.4.0 --conf spark.kubernetes.driver.annotation.sparkoperator.k8s.io/ownerreference=ChBTcGFya0FwcGxpY2F0aW9uGgpweXNwYXJrLXBpIiRmMjMyMzcxNS0yZmNjLTExZTktOWZhYi0wYTk0N2E1OWE1OWIqHHNwYXJrb3BlcmF0b3IuazhzLmlvL3YxYmV0YTE= s3a://riskflow-dev/spark-examples/pi.py]
I0213 20:21:29.486894 11 controller.go:92] Event(v1.ObjectReference{Kind:"SparkApplication", Namespace:"default", Name:"pyspark-pi", UID:"f2323715-2fcc-11e9-9fab-0a947a59a59b", APIVersion:"sparkoperator.k8s.io/v1beta1", ResourceVersion:"8276261", FieldPath:""}): type: 'Normal' reason: 'SparkApplicationAdded' SparkApplication pyspark-pi was added, Enqueuing it for submission
I0213 20:21:31.228686 11 controller.go:207] SparkApplication default/spark-pi was updated, enqueueing it
I0213 20:21:31.228989 11 controller.go:207] SparkApplication default/pyspark-pi was updated, enqueueing it
I0213 20:21:31.229019 11 controller.go:247] Starting processing key: "default/spark-pi"
W0213 20:21:31.229403 11 controller.go:323] driver not found for SparkApplication: default/spark-pi
I0213 20:21:31.229536 11 controller.go:509] Trying to update SparkApplication default/spark-pi, from: [{spark-eb23d02c938843b69965e4223ca20705 2019-02-13 03:21:47 +0000 UTC 0001-01-01 00:00:00 +0000 UTC {spark-pi-ui-svc 31882 169.55.168.180:31882 spark-pi-driver} {SUBMITTED } map[spark-pi-1550028105242-exec-1:FAILED] 1 1}] to [{spark-eb23d02c938843b69965e4223ca20705 2019-02-13 03:21:47 +0000 UTC 0001-01-01 00:00:00 +0000 UTC {spark-pi-ui-svc 31882 169.55.168.180:31882 spark-pi-driver} {SUBMITTED } map[spark-pi-1550028105242-exec-1:FAILED] 1 1}]
I0213 20:21:31.229887 11 controller.go:254] Ending processing key: "default/spark-pi"
E0213 20:21:31.745173 11 controller.go:569] failed to run spark-submit for SparkApplication default/pyspark-pi: failed to run spark-submit for SparkApplication default/pyspark-pi: Exception in thread "main" java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2195)
at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2654)
at org.apache.hadoop.fs.FileSystem.createFileSystem(FileSystem.java:2667)
at org.apache.hadoop.fs.FileSystem.access$200(FileSystem.java:94)
at org.apache.hadoop.fs.FileSystem$Cache.getInternal(FileSystem.java:2703)
at org.apache.hadoop.fs.FileSystem$Cache.get(FileSystem.java:2685)
at org.apache.hadoop.fs.FileSystem.get(FileSystem.java:373)
at org.apache.spark.deploy.DependencyUtils$.org$apache$spark$deploy$DependencyUtils$$resolveGlobPath(DependencyUtils.scala:190)
at org.apache.spark.deploy.DependencyUtils$$anonfun$resolveGlobPaths$2.apply(DependencyUtils.scala:146)
at org.apache.spark.deploy.DependencyUtils$$anonfun$resolveGlobPaths$2.apply(DependencyUtils.scala:144)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
at scala.collection.mutable.WrappedArray.foreach(WrappedArray.scala:35)
at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
at scala.collection.AbstractTraversable.flatMap(Traversable.scala:104)
at org.apache.spark.deploy.DependencyUtils$.resolveGlobPaths(DependencyUtils.scala:144)
at org.apache.spark.deploy.SparkSubmit$$anonfun$prepareSubmitEnvironment$4.apply(SparkSubmit.scala:355)
at org.apache.spark.deploy.SparkSubmit$$anonfun$prepareSubmitEnvironment$4.apply(SparkSubmit.scala:355)
at scala.Option.map(Option.scala:146)
at org.apache.spark.deploy.SparkSubmit.prepareSubmitEnvironment(SparkSubmit.scala:355)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:143)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:86)
at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:924)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:933)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
Caused by: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found
at org.apache.hadoop.conf.Configuration.getClassByName(Configuration.java:2101)
at org.apache.hadoop.conf.Configuration.getClass(Configuration.java:2193)
... 25 more

[rjanovski]

also encountered this issue.

I can't have an external jar/file from s3 placed in deps, because the operator tries to download the deps (not sure how) and fails.
This is especially bad, as I can't use the generic docker image and add the application jar as a dependency (at least not from s3).

[liyinan926]

It's not the operator binary that's downloading the dependencies. The error log you saw was from the Spark driver pod. The driver is responsible for downloading dependencies.

[olegfr]

@liyinan926 I don't think the error happens in the driver pod. This error happens in the Spark Operator code:

> kubectl logs queenly-starfish-sparkoperator-56766986-t2rfc --namespace spark-operator
...
E0213 20:21:31.745173 11 controller.go:569] failed to run spark-submit for SparkApplication default/pyspark-pi: failed to run spark-submit for SparkApplication default/pyspark-pi: Exception in thread "main" java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found

[liyinan926]

The error message you saw is the stderr output of executing the spark-submit command. The command itself failed and the error message was captured through stderr by the operator.

[olegfr]

@liyinan926 Sure, but the driver code then is executed inside Spark Operator pod that is missing classes to access S3 in $SPARK_HOME/jars directory. And this is the issue for me and others who try to use S3 as a cloud storage to run our Spark applications.

[liyinan926]

The operator simply runs the spark-submit command, which runs the submission client and launches the driver pod. The driver code runs in the driver Pod, not in the operator. The default Spark image doesn't have S3 dependencies built in.

[olegfr]

@liyinan926 Spark Operator fails to start the driver pod. This error happens before driver pod is started. I think that this error happens inside spark-submit in the Spark Operator pod.

[mihdih]

Hmm am not sure if this is expected, but as a work around, we can actually extend the base image and add the necessary jars as such:

FROM gcr.io/spark-operator/spark-operator:v2.4.0-v1beta1-latest

# add s3a connector
ADD http://central.maven.org/maven2/org/apache/hadoop/hadoop-aws/2.7.3/hadoop-aws-2.7.3.jar $SPARK_HOME/jars
ADD http://central.maven.org/maven2/com/amazonaws/aws-java-sdk/1.7.4.2/aws-java-sdk-1.7.4.2.jar $SPARK_HOME/jars

Hope it helps...

Update:

And build the image accordingly, which in turn will be used as the spark operator image

[rjanovski]

@mihdih instead of patching the operator image and using the deps, I find it easier to build my image with all its dependencies right in my spark project (build.sbt), adding:

• missing s3 access jars (for streaming checkpoints in my case, others may not need it anymore)
• prometheus
• my driver code in a fat-jar

dockerfile in docker := {
  // The assembly task generates a fat JAR file
  val artifact: File = assembly.value
  val artifactTargetPath = s"${name.value}.jar"

  new Dockerfile {
    from("gcr.io/spark-operator/spark:v2.4.0")
    addRaw("http://central.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.3.1/jmx_prometheus_javaagent-0.3.1.jar", "/prometheus/jmx_prometheus_javaagent.jar")
    addRaw("http://central.maven.org/maven2/org/apache/hadoop/hadoop-aws/2.7.3/hadoop-aws-2.7.3.jar", "$SPARK_HOME/jars")
    addRaw("http://central.maven.org/maven2/com/amazonaws/aws-java-sdk/1.7.4/aws-java-sdk-1.7.4.jar", "$SPARK_HOME/jars")

    add(artifact, artifactTargetPath)
  }
}

a slightly different workaround approach, hope this helps too :)

[mihdih]

@rjanovski Ahh yeah we still need that within our spark image, to support s3 connector. But in this case as what the reporter is pointing out, the spark-operator it self is prompting the missing class. Which puzzles me also, as i believe the pulling part should be done at the driver.

[liyinan926]

@mihdih yes the downloading of dependencies is handled by the driver. The error logs reported above shows that it's a Java exception thrown by the k8s submission client called by spark-submit. This error seems caused by the submission client failing to download the main app file that is also on S3 because of the missing aws dependencies. And the Java exception clearly indicates that. The error was output to stderr by spark-submit, and was captured by the operator that executed spark-submit. The operator is written purely in Golang so itself will never throw that kind of exception.

[mihdih]

@liyinan926 Ah I see. Thanks for the explanation. Totally makes sense now, I didn't notice that the main app is also on S3 🤦‍♂️

[liyinan926]

Closing this as this is expected behavior. To summarize:

• To be able to use s3a dependencies in applicationMainFile, you need to build a custom operator image with the aws bits, like what @mihdih did. Remote applicationMainFile is downloaded by the k8s submission client called by spark-submit. If the download fails, the submission fails and you will see error like the one reported above.
• To be able to use s3a dependencies in deps, you need to use a Spark image that has the aws bits. Remote dependencies specified this way are downloaded by the driver running in the driver pod. If the download fails, the driver fails.

[rjanovski]

@liyinan926 , this is a common scenario, and the remedy is just two more jars to add to the default operator image.

Why not just add them in, so you have the deps working also from s3 out-of-box?

[olegfr]

Let me to reiterate why I think the issue is with the spark operator, and not with my image. My application succeeds if I submit it the way below. Which in my opinion means that my image has all required jars to access s3.

mainApplicationFile: s3a://riskflow-dev/spark-examples/pi.py

However, if I add s3 file to deps.files, then I get the ClassNotFoundException:

mainApplicationFile: s3a://riskflow-dev/spark-examples/pi.py
deps:
files:

• s3a://riskflow-dev/spark-examples/rfstorage.py

Did anybody succeed to run SparkOperator with s3 files in deps.files?

[liyinan926]

Files/jars under dep are downloaded by the driver running in the driver pod. This means your Spark image must have the aws jars to be able to download dependencies from s3. mainApplicationFile is different as it's downloaded when spark-submit is run by the operator so the operator image must have the aws dependencies to support that.

[edencivitas]

I can't seem to get it to work with a custom Spark image. I tried the following configuration:

  mainApplicationFile: "local:///tmp/myjarname.jar"
  deps:
    jarsDownloadDir: "local:///tmp/"
    jars:
      - "s3a://somelocation/myjarname.jar"

I also created my own custom spark image with the aws-sdk and hadoop jars in /opt/spark/jobs and yet I still get the error that it can't find the S3 class.

driver:
    ...
    image: myCustomSparkImage

Within spark image docker container

bash-4.4# ls /opt/spark/jars | grep aws
aws-java-sdk-1.11.519.jar
hadoop-aws-2.8.5.jar

I'm still getting this error.

spark-operator  failed to submit SparkApplication myjar-test: failed to run spark-submit for SparkApplication namespace/myjar-test: Exception in thread "main" java.lang.RuntimeException: java.lang.ClassNotFoundException: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found

any ideas @liyinan926?

[runzhliu]

@edencivitas I think it's not easy to debug the spark-submit process in the operator pod, try to set the logLevel as 5 and to see what was going on while submitting your sparkapp, it might help.

https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/blob/master/pkg/controller/sparkapplication/submission.go#L65

[jrj-d]

Files/jars under dep are downloaded by the driver running in the driver pod. This means your Spark image must have the aws jars to be able to download dependencies from s3. mainApplicationFile is different as it's downloaded when spark-submit is run by the operator so the operator image must have the aws dependencies to support that.

@liyinan926 are you sure this is the case?
I'm working with code hosted on GCS (not S3), and from my experiments:

• having a mainApplicationFile on GCS requires adding the GCS connector to the Spark image used by the driver and the execs
• having a deps->pyFiles on GCS requires building the Spark operator with a base Spark image that contains the GCS connector

[barrydobson]

I know this is an old issue, but it's still the case with the latest operator that you need to rebuild it with the AWS/S3 jar files in order to be able to submit a job to your custom Spark container.

[liyinan926]

To clarify, for any files, jars, and pyFiles listed under dep that are remotely in S3 or GCS, the file URIs (e.g., s3a:// and gs:// ones) need to be resolved at submission time, and requires the jars implementing the file system to be present in the operator image, because currently the operator handles the submission by running spark-submit directly. mainApplicationFile is different, and using a mainApplicationFile in S3 or GCS does not requires the operator to be aware of the filesystem implementation.

[barrydobson]

The docs probably need to be updated in this case, and this statement is the opposite from the comments above. The user guide suggests that files or jars under deps can be from remote locations such as S3 or GS but doesn't mention anywhere that you would need to add dependancies to the operator image.

Thanks for clearing up

[liyinan926]

The docs probably need to be updated in this case, and this statement is the opposite from the comments above. The user guide suggests that files or jars under deps can be from remote locations such as S3 or GS but doesn't mention anywhere that you would need to add dependancies to the operator image.

We will be moving towards a model that uses a K8S Job of a single pod to run the submission, and have the operator manage the lifecycle of the Job. By default, the Job will use the same image of the driver, so as long as the driver image has S3 or GCS support built in, using remote dependencies on S3 or GCS just works. This is currently incubatorred in this branch: https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/tree/multi-version-support.

[thealmightygrant]

This will be a change to which service accounts need access to GCS or S3 even with the above mentioned change.

Also, it doesn't really matter what model this repo is changing to. People are using this operator right now, and they are expecting loading from GCS/S3 to work based on the docs. We should update the docs in order to let people know that this loading from GCS/S3 will not work with the current version. This will save people lots of pain when working with the current codebase.

[rehantahir]

What will be the version of jars and their names to use with the latest spark operator (gcr.io/spark-operator/spark-operator:latest)

@mihdih : I used your Dockerfile to build the custom spark operator, but ran into the following error:

• [[ -z root:x:0:0:root:/root:/bin/ash ]]
• exec /sbin/tini -s -- /usr/bin/spark-operator -v=2 -logtostderr -namespace= -enable-ui-service=true -ingress-url-format= -controller-threads=10 -resync-interval=30 -enable-batch-scheduler=false -label-selector-filter= -enable-metrics=true -metrics-labels=app_type -metrics-port=10254 -metrics-endpoint=/metrics -metrics-prefix= -enable-resource-quota-enforcement=false
  flag provided but not defined: -enable-ui-service
  Usage of /usr/bin/spark-operator:
  -alsologtostderr
  log to standard error as well as files........