scala - 如何让 Spark 2.0.0 SparkContext 和 SparkSql 函数在 scalatest/junit 测试环境中工作？

Question

自从迁移到 Spark 2.0.0 后，一些 SparkContext 和 SparkSql 函数(例如 textFile、parallelize、read.load)不再有效在 Scala 单元测试环境中，尽管它们确实在我的主类中的测试环境之外工作。当代码尝试在测试环境中执行这些函数中的每一个时，会发生 java.lang.ExceptionInInitializerError。请注意，无论我使用 SparkSession 还是旧的 SparkContext 和 SparkSql，在使用 Spark 2.0.0 时都存在同样的问题。此外，重要的是要注意，当我使用 Spark 1.6 时，相同的功能在测试环境中确实有效。解决这个问题的任何想法？测试代码和堆栈跟踪(与执行相关spark.sparkContext.textFile) 错误如下:

import org.apache.spark.sql.SparkSession
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import org.scalatest.{BeforeAndAfterAll, FunSuite}

@RunWith(classOf[JUnitRunner])
class TestSparkFunctions extends FunSuite with BeforeAndAfterAll {
  val master = "local[2]"
  val appName = "test"
  val spark = SparkSession
    .builder()
    .appName(appName)
    .master(master)
    .config("spark.sql.warehouse.dir", System.getProperty("user.dir"))
    .getOrCreate()
  val textFile = spark.sparkContext.textFile("people.txt")

  test("test spark functions") {

    //    val peopleDf = spark.read.format("org.apache.spark.sql.execution.datasources.csv.CSVFileFormat")
    //      .option("header", "true")
    //      .option("delimiter", "\\t")
    //      .option("mode", "PERMISSIVE")
    //      .option("inferSchema", "true")
    //       .load(getClass.getResource("/people.txt").getPath)

    //    val lines = Source.fromFile(getClass.getResource("/people.txt").getPath).getLines.toSeq
    //    val linesRdd: RDD[String] = spark.sparkContext.parallelize(lines)
  }
  override def afterAll() = {
    if (spark != null) spark.stop()
  }
}

堆栈跟踪错误:

java.lang.ExceptionInInitializerError
at org.apache.spark.SparkContext.withScope(SparkContext.scala:682)
at org.apache.spark.SparkContext.textFile(SparkContext.scala:800)
at com.corelogic.TestSparkFunctions.<init>(TestSparkFunctions.scala:19)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
at java.lang.Class.newInstance(Class.java:442)
at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:468)
at sbt.TestRunner.runTest$1(TestFramework.scala:76)
at sbt.TestRunner.run(TestFramework.scala:85)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1$$anonfun$apply$8.apply(TestFramework.scala:202)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1$$anonfun$apply$8.apply(TestFramework.scala:202)
at sbt.TestFramework$.sbt$TestFramework$$withContextLoader(TestFramework.scala:185)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1.apply(TestFramework.scala:202)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1.apply(TestFramework.scala:202)
at sbt.TestFunction.apply(TestFramework.scala:207)
at sbt.Tests$$anonfun$9.apply(Tests.scala:216)
at sbt.Tests$$anonfun$9.apply(Tests.scala:216)
at sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:44)
at sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:44)
at sbt.std.Transform$$anon$4.work(System.scala:63)
at sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:226)
at sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:226)
at sbt.ErrorHandling$.wideConvert(ErrorHandling.scala:17)
at sbt.Execute.work(Execute.scala:235)
at sbt.Execute$$anonfun$submit$1.apply(Execute.scala:226)
at sbt.Execute$$anonfun$submit$1.apply(Execute.scala:226)
at sbt.ConcurrentRestrictions$$anon$4$$anonfun$1.apply(ConcurrentRestrictions.scala:159)
at sbt.CompletionService$$anon$2.call(CompletionService.scala:28)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
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:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: com.fasterxml.jackson.databind.JsonMappingException: Jackson version is too old 2.4.4
at com.fasterxml.jackson.module.scala.JacksonModule$class.setupModule(JacksonModule.scala:56)
at com.fasterxml.jackson.module.scala.DefaultScalaModule.setupModule(DefaultScalaModule.scala:19)
at com.fasterxml.jackson.databind.ObjectMapper.registerModule(ObjectMapper.java:549)
at org.apache.spark.rdd.RDDOperationScope$.<init>(RDDOperationScope.scala:82)
at org.apache.spark.rdd.RDDOperationScope$.<clinit>(RDDOperationScope.scala)
at org.apache.spark.SparkContext.withScope(SparkContext.scala:682)
at org.apache.spark.SparkContext.textFile(SparkContext.scala:800)
at com.corelogic.TestSparkFunctions.<init>(TestSparkFunctions.scala:19)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
at java.lang.Class.newInstance(Class.java:442)
at org.scalatest.tools.Framework$ScalaTestTask.execute(Framework.scala:468)
at sbt.TestRunner.runTest$1(TestFramework.scala:76)
at sbt.TestRunner.run(TestFramework.scala:85)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1$$anonfun$apply$8.apply(TestFramework.scala:202)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1$$anonfun$apply$8.apply(TestFramework.scala:202)
at sbt.TestFramework$.sbt$TestFramework$$withContextLoader(TestFramework.scala:185)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1.apply(TestFramework.scala:202)
at sbt.TestFramework$$anon$2$$anonfun$$init$$1.apply(TestFramework.scala:202)
at sbt.TestFunction.apply(TestFramework.scala:207)
at sbt.Tests$$anonfun$9.apply(Tests.scala:216)
at sbt.Tests$$anonfun$9.apply(Tests.scala:216)
at sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:44)
at sbt.std.Transform$$anon$3$$anonfun$apply$2.apply(System.scala:44)
at sbt.std.Transform$$anon$4.work(System.scala:63)
at sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:226)
at sbt.Execute$$anonfun$submit$1$$anonfun$apply$1.apply(Execute.scala:226)
at sbt.ErrorHandling$.wideConvert(ErrorHandling.scala:17)
at sbt.Execute.work(Execute.scala:235)
at sbt.Execute$$anonfun$submit$1.apply(Execute.scala:226)
at sbt.Execute$$anonfun$submit$1.apply(Execute.scala:226)
at sbt.ConcurrentRestrictions$$anon$4$$anonfun$1.apply(ConcurrentRestrictions.scala:159)
at sbt.CompletionService$$anon$2.call(CompletionService.scala:28)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
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:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)

Answer 1

我遇到了同样的问题。这里的根本问题是:

Jackson version is too old 2.4.4

通过使用 Spark 2.x 支持的更新版本“2.6.5”解决了这个问题。 "com.fasterxml.jackson.core"% "jackson-databind"% "2.6.5"

引用:spark pom.xml