Estoy tratando de enviar mi aplicación Pyspark a un clúster de Kubernetes (Minikube) usando spark-submit:
./bin/spark-submit \ --master k8s://https://192.168.64.4:8443 \ --deploy-mode cluster \ --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1 \ --conf spark.kubernetes.container.image='pyspark:dev' \ --conf spark.kubernetes.container.image.pullPolicy='Never' \ local:///main.pyLa aplicación intenta llegar a una instancia de Kafka implementada dentro del clúster, por lo que especifiqué la dependencia del jar:
--packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.1La imagen del contenedor que estoy usando se basa en la que he creado con el script de utilidad. He incluido todas las dependencias de Python que necesita mi aplicación.
El controlador se implementa correctamente y obtiene el paquete Kafka (puedo proporcionar los registros si es necesario) y ejecuta el ejecutor en un nuevo módulo.
Pero luego el pod del ejecutor falla:
ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 1) java.lang.ClassNotFoundException: org.apache.spark.sql.kafka010.KafkaBatchInputPartition at java.net.URLClassLoader.findClass(URLClassLoader.java:382) at java.lang.ClassLoader.loadClass(ClassLoader.java:418) at java.lang.ClassLoader.loadClass(ClassLoader.java:351) at java.lang.Class.forName0(Native Method) at java.lang.Class.forName(Class.java:348) at org.apache.spark.serializer.JavaDeserializationStream$$anon$1.resolveClass(JavaSerializer.scala:68) at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1986) at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1850) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2160) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667) at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2405) at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2329) at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2187) at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1667) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:503) at java.io.ObjectInputStream.readObject(ObjectInputStream.java:461) at org.apache.spark.serializer.JavaDeserializationStream.readObject(JavaSerializer.scala:76) at org.apache.spark.serializer.JavaSerializerInstance.deserialize(JavaSerializer.scala:115) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:407) 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:748)Así que hice mis investigaciones en el pod ejecutor y descubrí que el jar no está presente (como se indica en el seguimiento de la pila) en la carpeta $SPARK_CLASSPATH (que está configurada en ':/opt/spark/jars/*')
¿Debo obtener e incluir también la dependencia en la carpeta Spark Jars al crear la imagen de la ventana acoplable? (Pensé que la opción '--packages' también haría que el ejecutor recuperara el jar especificado)
¿Comenzó con el Dockerfile oficial (kubernetes/dockerfiles/spark/bindings/python/Dockerfile) como se describe en la sección de imágenes de Docker de la documentación? También debe especificar una ubicación de carga en un sistema de archivos compatible con Hadoop y asegurarse de que los directorios de caché e inicio de Ivy especificados tengan los permisos correctos, como se describe en la sección Administración de dependencias .
Ejemplo de los documentos:
... --packages com.amazonaws:aws-java-sdk:1.7.4,org.apache.hadoop:hadoop-aws:2.7.6 --conf spark.kubernetes.file.upload.path=s3a://<s3-bucket>/path --conf spark.hadoop.fs.s3a.access.key=... --conf spark.hadoop.fs.s3a.impl=org.apache.hadoop.fs.s3a.S3AFileSystem --conf spark.hadoop.fs.s3a.fast.upload=true --conf spark.hadoop.fs.s3a.secret.key=.... --conf spark.driver.extraJavaOptions="-Divy.cache.dir=/tmp -Divy.home=/tmp" ...