如何使用Scala模拟Spark DataFrameReader？

Question

代码示例：

object ConfigurationLoader {

def readTable(tableName: String)(implicit spark: SparkSession): DataFrame = {
    spark.read
      .format("jdbc")
      .option("url", s"$postgresUrl/$postgresDatabase")
      .option("dbtable", tableName)
      .option("user", postgresUsername)
      .option("password", postgresPassword)
      .option("driver", postgresDriver)
      .load()
  }

def loadUsingFilter(dummyFilter: String*)(implicit spark: SparkSession): DataFrame = {
    readTable(postgresFilesTableName)
      .where(col("column").isin(fileTypes: _*))
  }
}

第二个问题--模拟scala对象，似乎需要使用其他方法来创建这样的服务。

Answer 1

在我看来，单元测试并不意味着测试数据库连接。这应该在集成测试中完成，以检查所有部分是否协同工作。单元测试只是用来测试您的功能逻辑，而不是Spark从数据库中读取的能力。

这就是为什么我会以稍微不同的方式设计您的代码，并且这样做，而不关心DB。

/** This, I don't test. I trust spark.read */
def readTable(tableName: String)(implicit spark: SparkSession): DataFrame = {
    spark.read
    .option(...)
    ...
    .load()
    // Nothing more
}

/** This I test, this is my logic. */
def transform(df : DataFrame, dummyFilter: String*): DataFrame = {
    df
      .where(col("column").isin(fileTypes: _*))
}

然后我在生产中以这种方式使用代码。

val source = readTable("...")
val result = transform(source, filter)

val df = Seq((1, Some("a"), true), (2, Some("b"), false), 
      (3, None, true)).toDF("x", "y", "z")
// and the test
val result = transform(df, filter)
result should be ...