Amazon Deequ在K8s容器运行失败但本地正常的问题求助

问题背景 #

使用Deequ版本2.0.0-spark-3.1，本地Spark 3.1.1环境运行测试代码可正常输出约束建议，但在K8s集群中运行相同代码时抛出java.lang.NoSuchMethodError错误，错误栈指向AggregateFunction.toAggregateExpression(boolean)方法缺失。

测试代码 #

val df = Seq(1,2,3).toDF("id")

df.show()

val suggestionsResult = ConstraintSuggestionRunner()
  .onData(df)
  .addConstraintRules(Rules.DEFAULT)
  .run

suggestionsResult.constraintSuggestions.foreach {
  case (column, suggestions) => suggestions.foreach { suggestion =>
      println(
        s"Constraint suggestion for $column: \t${suggestion.description}\n" +
          s"the corresponding scala code is ${suggestion.codeForConstraint}"
      )
    }
}

报错信息 #

java.lang.NoSuchMethodError: 'org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression org.apache.spark.sql.catalyst.expressions.aggregate.AggregateFunction.toAggregateExpression(boolean)'
    at org.apache.spark.sql.DeequFunctions$.withAggregateFunction(DeequFunctions.scala:31)
    at org.apache.spark.sql.DeequFunctions$.stateful_approx_count_distinct(DeequFunctions.scala:60)
    at com.amazon.deequ.analyzers.ApproxCountDistinct.aggregationFunctions(ApproxCountDistinct.scala:52)
    at com.amazon.deequ.analyzers.runners.AnalysisRunner$.$anonfun$runScanningAnalyzers$3(AnalysisRunner.scala:319)
    at scala.collection.TraversableLike.$anonfun$flatMap$1(TraversableLike.scala:245)
    at scala.collection.immutable.List.foreach(List.scala:392)
    at scala.collection.TraversableLike.flatMap(TraversableLike.scala:245)
    at scala.collection.TraversableLike.flatMap$(TraversableLike.scala:242)
    at scala.collection.immutable.List.flatMap(List.scala:355)
    at com.amazon.deequ.analyzers.runners.AnalysisRunner$.liftedTree1$1(AnalysisRunner.scala:319)
    at com.amazon.deequ.analyzers.runners.AnalysisRunner$.runScanningAnalyzers(AnalysisRunner.scala:318)
    at com.amazon.deequ.analyzers.runners.AnalysisRunner$.doAnalysisRun(AnalysisRunner.scala:167)
    at com.amazon.deequ.analyzers.runners.AnalysisRunBuilder.run(AnalysisRunBuilder.scala:110)
    at com.amazon.deequ.profiles.ColumnProfiler$.profile(ColumnProfiler.scala:141)
    at com.amazon.deequ.profiles.ColumnProfilerRunner.run(ColumnProfilerRunner.scala:72)
    at com.amazon.deequ.profiles.ColumnProfilerRunBuilder.run(ColumnProfilerRunBuilder.scala:185)
    at com.amazon.deequ.suggestions.ConstraintSuggestionRunner.profileAndSuggest(ConstraintSuggestionRunner.scala:203)
    at com.amazon.deequ.suggestions.ConstraintSuggestionRunner.run(ConstraintSuggestionRunner.scala:102)
    at com.amazon.deequ.suggestions.ConstraintSuggestionRunBuilder.run(ConstraintSuggestionRunBuilder.scala:226)

本地和K8s镜像均标注为Spark 3.1.1，观察Deequ的jar包发现包含org.apache.spark目录，疑似类加载冲突导致方法找不到。

问题原因 #

Deequ的2.0.0-spark-3.1版本内部打包了部分Spark类，这些类与K8s集群环境中Spark提供的系统类存在兼容性问题：

• 本地环境中，项目依赖的Spark类加载优先级高于Deequ打包的类，因此不会触发冲突；
• K8s集群模式下，Spark系统类加载器优先级更高，Deequ打包的类调用的方法签名与集群Spark类不匹配，导致NoSuchMethodError。

解决办法 #

1. 排除Deequ中的Spark依赖 #

修改项目依赖配置，强制使用集群环境的Spark类，排除Deequ自带的Spark模块：

• SBT配置：

libraryDependencies += "com.amazon.deequ" % "deequ" % "2.0.0-spark-3.1" excludeAll(
  ExclusionRule(organization = "org.apache.spark")
)

• Maven配置：

<dependency>
  <groupId>com.amazon.deequ</groupId>
  <artifactId>deequ</artifactId>
  <version>2.0.0-spark-3.1</version>
  <exclusions>
    <exclusion>
      <groupId>org.apache.spark</groupId>
      <artifactId>*</artifactId>
    </exclusion>
  </exclusions>
</dependency>

2. 调整Spark作业类加载顺序 #

提交Spark作业到K8s时，添加以下参数让用户依赖类优先加载：

spark-submit --conf spark.driver.userClassPathFirst=true --conf spark.executor.userClassPathFirst=true ...

注意：该参数可能引发其他类冲突，需测试验证。

3. 更换兼容的Deequ版本 #

尝试升级Deequ版本至2.0.1-spark-3.1或其他与Spark 3.1.1适配的版本，新版本可能修复了类打包的兼容性问题。

4. 验证K8s镜像的Spark版本 #

进入K8s容器执行spark-submit --version，确认镜像实际使用的Spark版本与标注一致，避免版本标签与实际内容不符。

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