scala - 对于基本数据框创建示例，我应该如何在 Spark 中编写单元测试？

Question

我正在努力使用 Spark 提供的示例文本文件编写用于创建数据框的基本单元测试，如下所示。

class dataLoadTest extends FunSuite with Matchers with BeforeAndAfterEach {

private val master = "local[*]"
private val appName = "data_load_testing"

private var spark: SparkSession = _

override def beforeEach() {
  spark = new SparkSession.Builder().appName(appName).getOrCreate()
}

import spark.implicits._

 case class Person(name: String, age: Int)

  val df = spark.sparkContext
      .textFile("/Applications/spark-2.2.0-bin-hadoop2.7/examples/src/main/resources/people.txt")
      .map(_.split(","))
      .map(attributes => Person(attributes(0),attributes(1).trim.toInt))
      .toDF()

  test("Creating dataframe should produce data from of correct size") {
  assert(df.count() == 3)
  assert(df.take(1).equals(Array("Michael",29)))
}

override def afterEach(): Unit = {
  spark.stop()
}

}

我知道代码本身可以工作(从spark.implicits._ .... toDF())，因为我已经在Spark-Scala shell中验证了这一点，但在测试类中我收到了很多错误； IDE 无法识别“import Spark.implicits._”或“toDF()”，因此测试不会运行。

我正在使用 SparkSession，它会在后台自动创建 SparkConf、SparkContext 和 SQLContext。

我的代码仅使用 Spark 存储库中的示例代码。

有什么想法为什么这不起作用吗？谢谢!

注意。我已经在 StackOverflow 上查看过 Spark 单元测试问题，如下所示:How to write unit tests in Spark 2.0+?我已经用它来编写测试，但仍然收到错误。

我正在使用 Scala 2.11.8 和 Spark 2.2.0 以及 SBT 和 IntelliJ。这些依赖项已正确包含在 SBT 构建文件中。运行测试时出现的错误是:

错误:(29, 10) 值 toDF 不是 org.apache.spark.rdd.RDD[dataLoadTest.this.Person] 的成员可能的原因:也许“value toDF”之前缺少分号？ .toDF()

错误:(20, 20) 需要稳定标识符，但发现了 dataLoadTest.this.spark.implicits。 导入spark.implicits._

IntelliJ 无法识别 import spark.implicits._ 或 .toDF() 方法。

我导入了:导入 org.apache.spark.sql.SparkSession导入 org.scalatest.{BeforeAndAfterEach、FlatSpec、FunSuite、Matchers}

Answer 1

您需要将 sqlContext 分配给 val 才能使 implicits 正常工作。由于您的 sparkSession 是 var，因此 implicits 无法使用它

所以你需要这样做

val sQLContext = spark.sqlContext
import sQLContext.implicits._

此外，您可以为测试编写函数，以便您的测试类如下所示

    class dataLoadTest extends FunSuite with Matchers with BeforeAndAfterEach {

  private val master = "local[*]"
  private val appName = "data_load_testing"

  var spark: SparkSession = _

  override def beforeEach() {
    spark = new SparkSession.Builder().appName(appName).master(master).getOrCreate()
  }


  test("Creating dataframe should produce data from of correct size") {
    val sQLContext = spark.sqlContext
    import sQLContext.implicits._

    val df = spark.sparkContext
    .textFile("/Applications/spark-2.2.0-bin-hadoop2.7/examples/src/main/resources/people.txt")
    .map(_.split(","))
    .map(attributes => Person(attributes(0), attributes(1).trim.toInt))
    .toDF()

    assert(df.count() == 3)
    assert(df.take(1)(0)(0).equals("Michael"))
  }

  override def afterEach() {
    spark.stop()
  }

}
case class Person(name: String, age: Int)