相关疑难解决方法(0)

Scala in Depth演示了Loaner模式:

def readFile[T](f: File)(handler: FileInputStream => T): T = {
  val resource = new java.io.FileInputStream(f)
  try {
    handler(resource)
  } finally {
      resource.close()
  }
}

用法示例:

readFile(new java.io.File("test.txt")) { input =>
   println(input.readByte)
}

此代码看似简单明了.什么是Scala中Loaner模式的"反模式",以便我知道如何避免它？

我正在尝试测试我的程序的一部分,它执行数据帧的转换我想测试这些数据帧的几个不同变体,这排除了从文件中读取特定DF的选项

所以我的问题是:

1. 有没有关于如何使用Spark和数据帧执行单元测试的优秀教程,特别是关于数据帧的创建？
2. 如何在没有大量样板的情况下创建这些不同的几行数据帧,而无需从文件中读取这些数据帧？
3. 是否有任何实用程序类用于检查数据框内的特定值？

我之前显然用谷歌搜索过,但找不到任何非常有用的东西.我找到的更有用的链接包括:

• 使用数据框运行基本单元测试
• 使用DF定制断言

如果示例/教程在Scala中会很棒,但我会采用你所拥有的任何语言

提前致谢

试图测试Spark Structured Streams ......并且失败......我该如何正确测试它们？

我从这里跟踪了一般的Spark测试问题,我最接近的尝试是[ 1 ]看起来像:

import simpleSparkTest.SparkSessionTestWrapper
import org.scalatest.FunSpec  
import org.apache.spark.sql.types.{StringType, IntegerType, DoubleType, StructType, DateType}
import org.apache.spark.sql.streaming.OutputMode

class StructuredStreamingSpec extends FunSpec with SparkSessionTestWrapper {

  describe("Structured Streaming") {

    it("Read file from system") {

      val schema = new StructType()
        .add("station_id", IntegerType)
        .add("name", StringType)
        .add("lat", DoubleType)
        .add("long", DoubleType)
        .add("dockcount", IntegerType)
        .add("landmark", StringType)
        .add("installation", DateType)

      val sourceDF = spark.readStream
        .option("header", "true")
        .schema(schema)
        .csv("/Spark-The-Definitive-Guide/data/bike-data/201508_station_data.csv")
        .coalesce(1)

      val countSource = sourceDF.count()

      val query = sourceDF.writeStream
        .format("memory")
        .queryName("Output")
        .outputMode(OutputMode.Append())
        .start()
        .processAllAvailable()

      assert(countSource === 70) …

我的 Spark 应用程序中有一个方法可以从 MySQL 数据库加载数据。该方法看起来像这样。

trait DataManager {

val session: SparkSession

def loadFromDatabase(input: Input): DataFrame = {
            session.read.jdbc(input.jdbcUrl, s"(${input.selectQuery}) T0",
              input.columnName, 0L, input.maxId, input.parallelism, input.connectionProperties)
    }
}

该方法除了执行jdbc方法并从数据库加载数据之外不执行任何其他操作。我该如何测试这个方法？标准方法是创建对象的模拟，session该对象是 的实例SparkSession。但由于SparkSession有一个私有构造函数，我无法使用 ScalaMock 来模拟它。

这里的主要问题是我的函数是一个纯粹的副作用函数（副作用是从关系数据库中提取数据），并且鉴于我在模拟时遇到问题，我如何对该函数进行单元测试SparkSession。

那么有什么方法可以模拟SparkSession或者比模拟更好的方法来测试这个方法呢？

我正在努力编写一个基本单元测试来创建数据框,使用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 …