We are working on a project where we have to deploy our application on a Spark cluster( based upon EKS). We are using Spark-Operator to manage our Spark cluster.
Application Nature: My application is based on Spark's "structured streaming". It streams events from the Kafka topic with 5 partitions.
The application is almost get deployed using the help from AWS for Spark Operator (Link) but it is not able to start the executors. It gets exit with the below error:
23/07/10 12:50:16 INFO Executor: Fetching file://usr/lib/jars/java-word-count.jar with timestamp 1688993383773
23/07/10 12:50:16 ERROR CoarseGrainedExecutorBackend: Executor self-exiting due to : Unable to create executor due to URI has an authority component
java.lang.IllegalArgumentException: URI has an authority component
at java.io.File.<init>(File.java:425)
at org.apache.spark.util.Utils$.doFetchFile(Utils.scala:778)
at org.apache.spark.util.Utils$.fetchFile(Utils.scala:537)
at org.apache.spark.executor.Executor.$anonfun$updateDependencies$13(Executor.scala:962)
at org.apache.spark.executor.Executor.$anonfun$updateDependencies$13$adapted(Executor.scala:954)
at scala.collection.TraversableLike$WithFilter.$anonfun$foreach$1(TraversableLike.scala:985)
at scala.collection.mutable.HashMap.$anonfun$foreach$1(HashMap.scala:149)
at scala.collection.mutable.HashTable.foreachEntry(HashTable.scala:237)
at scala.collection.mutable.HashTable.foreachEntry$(HashTable.scala:230)
at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:44)
at scala.collection.mutable.HashMap.foreach(HashMap.scala:149)
at scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:984)
at org.apache.spark.executor.Executor.org$apache$spark$executor$Executor$$updateDependencies(Executor.scala:954)
at org.apache.spark.executor.Executor.<init>(Executor.scala:247)
at org.apache.spark.executor.CoarseGrainedExecutorBackend$$anonfun$receive$1.applyOrElse(CoarseGrainedExecutorBackend.scala:185)
at org.apache.spark.rpc.netty.Inbox.$anonfun$process$1(Inbox.scala:115)
at org.apache.spark.rpc.netty.Inbox.safelyCall(Inbox.scala:213)
at org.apache.spark.rpc.netty.Inbox.process(Inbox.scala:100)
at org.apache.spark.rpc.netty.MessageLoop.org$apache$spark$rpc$netty$MessageLoop$$receiveLoop(MessageLoop.scala:75)
at org.apache.spark.rpc.netty.MessageLoop$$anon$1.run(MessageLoop.scala:41)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:750)
23/07/10 12:50:16 INFO CoarseGrainedExecutorBackend: Driver commanded a shutdown
23/07/10 12:50:16 INFO MemoryStore: MemoryStore cleared
23/07/10 12:50:16 INFO BlockManager: BlockManager stopped
And pod yaml is :
apiVersion: "sparkoperator.k8s.io/v1beta2"
kind: SparkApplication
metadata:
name: spark-mango
namespace: spark-operator
spec:
type: Java
mode: cluster
#image: "755674844232.dkr.ecr.us-east-1.amazonaws.com/spark/emr-6.11.0:latest"
image: "603016229198.dkr.ecr.us-east-1.amazonaws.com/emr6.geomango_spark_cluster"
imagePullPolicy: Always
mainClass: com.ageon.geomango.WordCount
arguments:
- "b-4.geomangoemrdemo.pzs37h.c12.kafka.us-east-1.amazonaws.com:9092,b-3.geomangoemrdemo.pzs37h.c12.kafka.us-east-1.amazonaws.com:9092,b-9.geomangoemrdemo.pzs37h.c12.kafka.us-east-1.amazonaws.com:9092"
mainApplicationFile: "local:///usr/lib/jars/java-word-count.jar"
sparkVersion: "3.3.1"
hadoopConf:
# EMRFS filesystem
fs.s3.customAWSCredentialsProvider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
fs.s3.impl: com.amazon.ws.emr.hadoop.fs.EmrFileSystem
fs.AbstractFileSystem.s3.impl: org.apache.hadoop.fs.s3.EMRFSDelegate
fs.s3.buffer.dir: /mnt/s3
fs.s3.getObject.initialSocketTimeoutMilliseconds: "2000"
mapreduce.fileoutputcommitter.algorithm.version.emr_internal_use_only.EmrFileSystem: "2"
mapreduce.fileoutputcommitter.cleanup-failures.ignored.emr_internal_use_only.EmrFileSystem: "true"
sparkConf:
# Required for EMR Runtime
spark.driver.extraClassPath: /usr/lib/hadoop-lzo/lib/*:/usr/lib/hadoop/hadoop-aws.jar:/usr/share/aws/aws-java-sdk/*:/usr/share/aws/emr/emrfs/conf:/usr/share/aws/emr/emrfs/lib/*:/usr/share/aws/emr/emrfs/auxlib/*:/usr/share/aws/emr/security/conf:/usr/share/aws/emr/security/lib/*:/usr/share/aws/hmclient/lib/aws-glue-datacatalog-spark-client.jar:/usr/share/java/Hive-JSON-Serde/hive-openx-serde.jar:/usr/share/aws/sagemaker-spark-sdk/lib/sagemaker-spark-sdk.jar:/home/hadoop/extrajars/*
spark.driver.extraLibraryPath: /usr/lib/hadoop/lib/native:/usr/lib/hadoop-lzo/lib/native:/docker/usr/lib/hadoop/lib/native:/docker/usr/lib/hadoop-lzo/lib/native
spark.executor.extraClassPath: /usr/lib/hadoop-lzo/lib/*:/usr/lib/hadoop/hadoop-aws.jar:/usr/share/aws/aws-java-sdk/*:/usr/share/aws/emr/emrfs/conf:/usr/share/aws/emr/emrfs/lib/*:/usr/share/aws/emr/emrfs/auxlib/*:/usr/share/aws/emr/security/conf:/usr/share/aws/emr/security/lib/*:/usr/share/aws/hmclient/lib/aws-glue-datacatalog-spark-client.jar:/usr/share/java/Hive-JSON-Serde/hive-openx-serde.jar:/usr/share/aws/sagemaker-spark-sdk/lib/sagemaker-spark-sdk.jar:/home/hadoop/extrajars/*
spark.executor.extraLibraryPath: /usr/lib/hadoop/lib/native:/usr/lib/hadoop-lzo/lib/native:/docker/usr/lib/hadoop/lib/native:/docker/usr/lib/hadoop-lzo/lib/native
restartPolicy:
type: Never
volumes:
- name: efs-spark-operator
persistentVolumeClaim:
claimName: efs-storage-claim
- name: efs-spark-operator-executors
persistentVolumeClaim:
claimName: efs-storage-claim-executors
driver:
cores: 2
memory: "9g"
labels:
version: 3.3.2
nodeSelector:
emrtype: sf-emr-on-spot
serviceAccount: driver-account-sa
volumeMounts:
- name: efs-spark-operator
mountPath: /mnt1
executor:
cores: 2
instances: 1
memory: "9g"
nodeSelector:
emrtype: sf-emr-on-spot
labels:
version: 3.3.2
volumeMounts:
- name: efs-spark-operator-executors
mountPath: /mnt1
Any help will be much appreciated.