ClassNotFoundException：spark 提交期间 org.apache.beam.runners.spark.io.SourceRDD$SourcePartition

Question

我使用

spark-submit

来触发独立集群来执行我的着色 jar，但是执行器出现错误：

22/12/06 15:21:25 INFO TaskSetManager: Starting task 0.1 in stage 0.0 (TID 1) (10.37.2.77, executor 0, partition 0, PROCESS_LOCAL, 5133 bytes) taskResourceAssignments Map()
22/12/06 15:21:25 INFO TaskSetManager: Lost task 0.1 in stage 0.0 (TID 1) on 10.37.2.77, executor 0: java.lang.ClassNotFoundException (org.apache.beam.runners.spark.io.SourceRDD$SourcePartition) [duplicate 1]
22/12/06 15:21:25 INFO TaskSetManager: Starting task 0.2 in stage 0.0 (TID 2) (10.37.2.77, executor 0, partition 0, PROCESS_LOCAL, 5133 bytes) taskResourceAssignments Map()
22/12/06 15:21:25 INFO TaskSetManager: Lost task 0.2 in stage 0.0 (TID 2) on 10.37.2.77, executor 0: java.lang.ClassNotFoundException (org.apache.beam.runners.spark.io.SourceRDD$SourcePartition) [duplicate 2]
22/12/06 15:21:25 INFO TaskSetManager: Starting task 0.3 in stage 0.0 (TID 3) (10.37.2.77, executor 0, partition 0, PROCESS_LOCAL, 5133 bytes) taskResourceAssignments Map()
22/12/06 15:21:25 INFO TaskSetManager: Lost task 0.3 in stage 0.0 (TID 3) on 10.37.2.77, executor 0: java.lang.ClassNotFoundException (org.apache.beam.runners.spark.io.SourceRDD$SourcePartition) [duplicate 3]
22/12/06 15:21:25 ERROR TaskSetManager: Task 0 in stage 0.0 failed 4 times; aborting job
22/12/06 15:21:25 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool 
22/12/06 15:21:25 INFO TaskSchedulerImpl: Cancelling stage 0
22/12/06 15:21:25 INFO TaskSchedulerImpl: Killing all running tasks in stage 0: Stage cancelled
22/12/06 15:21:25 INFO DAGScheduler: ResultStage 0 (collect at BoundedDataset.java:96) failed in 1.380 s due to Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 3) (10.37.2.77 executor 0): java.lang.ClassNotFoundException: org.apache.beam.runners.spark.io.SourceRDD$SourcePartition
    at java.lang.ClassLoader.findClass(ClassLoader.java:523)
    at org.apache.spark.util.ParentClassLoader.findClass(ParentClassLoader.java:35)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
    at org.apache.spark.util.ParentClassLoader.loadClass(ParentClassLoader.java:40)
    at org.apache.spark.util.ChildFirstURLClassLoader.loadClass(ChildFirstURLClassLoader.java:48)
    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:1988)
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1852)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2186)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669)
    at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2431)
    at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2355)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2213)
    at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1669)
    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:458)
    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)

我的要求如下：

 curl -X POST http://xxxxxx:6066/v1/submissions/create --header "Content-Type:application/json;charset=UTF-8" --data '{
  "appResource": "/home/xxxx/xxxx-bundled-0.1.jar",
  "sparkProperties": {
    "spark.master": "spark://xxxxxxx:7077",
    "spark.driver.userClassPathFirst": "true",
    "spark.executor.userClassPathFirst": "true",
    "spark.app.name": "DataPipeline",
    "spark.submit.deployMode": "cluster",
    "spark.driver.supervise": "true"
  },
  "environmentVariables": {
    "SPARK_ENV_LOADED": "1"
  },
  "clientSparkVersion": "3.1.3",
  "mainClass": "com.xxxx.DataPipeline",
  "action": "CreateSubmissionRequest",
  "appArgs": [
    "--config=xxxx",
    "--runner=SparkRunner"
  ]

我设置了

"spark.driver.userClassPathFirst": "true"

和

"spark.executor.userClassPathFirst": "true"

，因为我的jar中使用了proto3。 不知道为什么在执行器上找不到这个类。 我的beam版本2.41.0，spark版本3.1.3，hadoop版本3.2.0。

最后，我将着色插件升级到 3.4.0，然后 protobuf 的重新定位工作正常，我删除了

"spark.driver.userClassPathFirst": "true"

和

"spark.executor.userClassPathFirst": "true"

。之后一切正常。

spark-submit

本地或通过rest api都可以。

Answer 1

最后，我将着色插件升级到 3.4.0，然后 protobuf 的重定位工作正常，并删除了 "spark.driver.userClassPathFirst": "true" 和 "spark.executor.userClassPathFirst": "true"。之后一切正常。本地“spark-submit”或通过rest api 都可以。

在着色插件中重新定位 protobuf：

<relocations>
    <relocation>
        <pattern>com.google.protobuf</pattern>
        <shadedPattern>shaded.com.google.protobuf</shadedPattern>
    </relocation>
</relocations>

然后以下两个调用都有效：

spark-submit --class XXXXPipeline --master spark://xxxx:7077 --deploy-mode cluster --supervise /xxxxx/xxxx-bundled-0.1.jar

和

curl -X POST http://xxxx:6066/v1/submissions/create --header "Content-Type:application/json;charset=UTF-8" --data '{
  "appResource": "xxxxx-bundled-0.1.jar",
  "sparkProperties": {
    "spark.master": "spark://XXXXXX:7077",
    "spark.jars": "xxxx-bundled-0.1.jar",
    "spark.app.name": "XXXPipeline",
    "spark.submit.deployMode": "cluster",
    "spark.driver.supervise": "true"
  },
  "environmentVariables": {
    "SPARK_ENV_LOADED": "1"
  },
  "clientSparkVersion": "3.1.3",
  "mainClass": "XXXXPipeline",
  "action": "CreateSubmissionRequest",
  "appArgs": []
}'