Dataproc Serverless 2.1版本PubSubLite Spark连接器兼容错误解决

关键补充信息 #

• 使用Dataproc Serverless镜像版本2.0可正常运行无报错，该问题是从--version 2.0升级到--version 2.1后的回归问题。

问题概述 #

通过以下命令提交Dataproc Serverless的PubSubLite相关PySpark批处理任务：

gcloud dataproc batches submit pyspark \
  ./main.py \
  --region europe-west1 \
  --deps-bucket gs://... \
  --jars gs://spark-lib/pubsublite/pubsublite-spark-sql-streaming-1.0.0-with-dependencies.jar \
  --version 2.1

执行后触发兼容性错误，具体报错信息如下：

23/10/04 14:02:23 ERROR MicroBatchExecution: Query [id = e1d920ae-c3fd-47ab-8139-474080e76a16, runId = 40153dad-ce4f-4d2a-ae7f-b860ab145ab2] terminated with error
java.lang.IncompatibleClassChangeError: Class com.google.longrunning.GetOperationRequest does not implement the requested interface repackaged.com.google.protobuf.MessageLite
        at com.google.cloud.pubsublite.repackaged.io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.<init>(ProtoLiteUtils.java:128)
        at com.google.cloud.pubsublite.repackaged.io.grpc.protobuf.lite.ProtoLiteUtils.marshaller(ProtoLiteUtils.java:84)
        at com.google.cloud.pubsublite.repackaged.io.grpc.protobuf.ProtoUtils.marshaller(ProtoUtils.java:57)
        at com.google.longrunning.stub.GrpcOperationsStub.<clinit>(GrpcOperationsStub.java:68)
        at com.google.cloud.pubsublite.v1.stub.GrpcCursorServiceStub.<init>(GrpcCursorServiceStub.java:130)
        at com.google.cloud.pubsublite.v1.stub.GrpcCursorServiceStub.<init>(GrpcCursorServiceStub.java:116)
        at com.google.cloud.pubsublite.v1.stub.GrpcCursorServiceStub.create(GrpcCursorServiceStub.java:96)
        at com.google.cloud.pubsublite.v1.stub.CursorServiceStubSettings.createStub(CursorServiceStubSettings.java:201)
        at com.google.cloud.pubsublite.v1.CursorServiceClient.<init>(CursorServiceClient.java:160)
        at com.google.cloud.pubsublite.v1.CursorServiceClient.create(CursorServiceClient.java:142)
        at com.google.cloud.pubsublite.spark.PslReadDataSourceOptions.newCursorServiceClient(PslReadDataSourceOptions.java:155)
        at com.google.cloud.pubsublite.spark.PslReadDataSourceOptions.newCursorClient(PslReadDataSourceOptions.java:169)
        at com.google.cloud.pubsublite.spark.PslMicroBatchStream.<init>(PslMicroBatchStream.java:59)
        at com.google.cloud.pubsublite.spark.PslScanBuilder.toMicroBatchStream(PslScanBuilder.java:110)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.$anonfun$applyOrElse$4(MicroBatchExecution.scala:107)
        at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:454)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:100)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:84)
        at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:104)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:488)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:456)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan$lzycompute(MicroBatchExecution.scala:84)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan(MicroBatchExecution.scala:64)
        at org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:298)
        at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827)
        at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:284)
        at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:207)
Exception in thread "stream execution thread for [id = e1d920ae-c3fd-47ab-8139-474080e76a16, runId = 40153dad-ce4f-4d2a-ae7f-b860ab145ab2]" java.lang.IncompatibleClassChangeError: Class com.google.longrunning.GetOperationRequest does not implement the requested interface repackaged.com.google.protobuf.MessageLite
        at com.google.cloud.pubsublite.repackaged.io.grpc.protobuf.lite.ProtoLiteUtils$MessageMarshaller.<init>(ProtoLiteUtils.java:128)
        at com.google.cloud.pubsublite.repackaged.io.grpc.protobuf.lite.ProtoLiteUtils.marshaller(ProtoLiteUtils.java:84)
        at com.google.cloud.pubsublite.repackaged.io.grpc.protobuf.ProtoUtils.marshaller(ProtoUtils.java:57)
        at com.google.longrunning.stub.GrpcOperationsStub.<clinit>(GrpcOperationsStub.java:68)
        at com.google.cloud.pubsublite.v1.stub.GrpcCursorServiceStub.<init>(GrpcCursorServiceStub.java:130)
        at com.google.cloud.pubsublite.v1.stub.GrpcCursorServiceStub.<init>(GrpcCursorServiceStub.java:116)
        at com.google.cloud.pubsublite.v1.stub.GrpcCursorServiceStub.create(GrpcCursorServiceStub.java:96)
        at com.google.cloud.pubsublite.v1.stub.CursorServiceStubSettings.createStub(CursorServiceStubSettings.java:201)
        at com.google.cloud.pubsublite.v1.CursorServiceClient.<init>(CursorServiceClient.java:160)
        at com.google.cloud.pubsublite.v1.CursorServiceClient.create(CursorServiceClient.java:142)
        at com.google.cloud.pubsublite.spark.PslReadDataSourceOptions.newCursorServiceClient(PslReadDataSourceOptions.java:155)
        at com.google.cloud.pubsublite.spark.PslReadDataSourceOptions.newCursorClient(PslReadDataSourceOptions.java:169)
        at com.google.cloud.pubsublite.spark.PslMicroBatchStream.<init>(PslMicroBatchStream.java:59)
        at com.google.cloud.pubsublite.spark.PslScanBuilder.toMicroBatchStream(PslScanBuilder.java:110)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.$anonfun$applyOrElse$4(MicroBatchExecution.scala:107)
        at scala.collection.mutable.HashMap.getOrElseUpdate(HashMap.scala:454)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:100)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution$$anonfun$1.applyOrElse(MicroBatchExecution.scala:84)
        at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:104)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:488)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transform(TreeNode.scala:456)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan$lzycompute(MicroBatchExecution.scala:84)
        at org.apache.spark.sql.execution.streaming.MicroBatchExecution.logicalPlan(MicroBatchExecution.scala:64)
        at org.apache.spark.sql.execution.streaming.StreamExecution.$anonfun$runStream$1(StreamExecution.scala:298)
        at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.scala:18)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827)
        at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runStream(StreamExecution.scala:284)
        at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:207)

问题分析 #

该错误属于Java依赖打包冲突问题：

• com.google.longrunning.stub.GrpcOperationsStub构造函数要求传入实现repackaged.com.google.protobuf.MessageLite接口的对象，该接口是PubSubLite依赖中重打包（shaded）的版本
• 但实际加载的com.google.longrunning.GetOperationRequest类实现的是非重打包的com.google.protobuf.MessageLite接口，导致类型不兼容
• 冲突依赖来自官方提供的gs://spark-lib/pubsublite/pubsublite-spark-sql-streaming-1.0.0-with-dependencies.jar，无法自行修改内部结构

关联Spark代码 #

from pyspark import sql as psql


spark = (
        psql.SparkSession.Builder()
        .getOrCreate()
    )

input_ = (
    spark.readStream.format("pubsublite")
        .option(
            "pubsublite.subscription",
            "...",
        )
        .load()
)

query = input_.writeStream.trigger(processingTime="2 seconds").format("console").start()

query.awaitTermination(10)

解决方案建议 #

1. 临时回退版本：将Dataproc Serverless版本改回2.0，修改提交命令中的--version参数为2.0，即可恢复正常运行
2. 升级依赖版本：检查PubSubLite Spark连接器是否有适配Dataproc 2.1的新版本，替换命令中的--jars路径为对应新版本的jar包
3. 反馈官方修复：向Google Cloud提交技术工单，说明该回归问题，推动官方发布兼容Dataproc 2.1的PubSubLite连接器版本

内容的提问来源于stack exchange，提问作者AlexisBRENON