使用数据框进行单元测试：收集返回空数组答案

【问题标题】：spark unit testing with dataframe : Collect return empty array使用数据框进行单元测试：收集返回空数组
【发布时间】：2015-11-07 09:32:40
【问题描述】：

我正在使用 spark，我一直在努力通过 Dataframe 和 Spark SQL 进行简单的单元测试。

这里是sn-p代码：

class TestDFSpec extends SharedSparkContext  { 
  "Test DF " should { 
    "pass equality" in { 
      val createDF = sqlCtx.createDataFrame(createsRDD,classOf[Test]).toDF() 
      createDF.registerTempTable("test") 

      sqlCtx.sql("select * FROM test").collectAsList() === List(Row(Test.from(create1)),Row(Test.from(create2))) 
    } 
  } 
  val create1 = "4869215,bbbbb" 
  val create2 = "4869215,aaaaa" 
  val createsRDD = sparkContext.parallelize(Seq(create1,create2)).map(Test.from) 
}

我从 spark github 复制代码并添加一些小的更改以提供 SQLContext：

trait SharedSparkContext extends Specification with BeforeAfterAll { 
  import net.lizeo.bi.spark.conf.JobConfiguration._ 

  @transient private var _sql: SQLContext = _ 

  def sqlCtx: SQLContext = _sql 

  override def beforeAll() { 

    println(sparkConf) 

    _sql = new SQLContext(sparkContext) 

  } 

  override def afterAll() { 
    sparkContext.stop() 
    _sql =  null 

  } 
}

模型测试非常简单：

case class Test(key:Int, value:String) 

  object Test { 
    def from(line:String):Test = { 
      val f = line.split(",") 
      Test(f(0).toInt,f(1)) 
    } 
  }

作业配置对象：

object JobConfiguration {
  val conf = ConfigFactory.load()

  val sparkName = conf.getString("spark.name")
  val sparkMaster = conf.getString("spark.master")

  lazy val sparkConf = new SparkConf()
    .setAppName(sparkName)
    .setMaster(sparkMaster)
    .set("spark.executor.memory",conf.getString("spark.executor.memory"))         
    .set("spark.io.compression.codec",conf.getString("spark.io.compression.codec"))

  val sparkContext = new SparkContext(sparkConf)  
}

我正在使用带有 Spec2 的 Spark 1.3.0。我的 sbt 项目文件的确切依赖项是：

object Dependencies { 
  private val sparkVersion = "1.3.0" 
  private val clouderaVersion = "5.4.4" 

  private val sparkClouderaVersion = s"$sparkVersion-cdh$clouderaVersion" 

  val sparkCdhDependencies = Seq( 
    "org.apache.spark" %% "spark-core" % sparkClouderaVersion % "provided", 
    "org.apache.spark" %% "spark-sql" % sparkClouderaVersion % "provided" 
    ) 

}

测试输出为：

[info] TestDFSpec  
[info]  
[info] Test DF  should  
[error]   x pass equality  
[error]    '[[], []]'  
[error]  
[error]     is not equal to  
[error]  
[error]    List([Test(4869215,bbbbb)], [Test(4869215,aaaaa)]) (TestDFSpec.scala:17)  
[error] Actual:   [[], []]  [error] Expected: List([Test(4869215,bbbbb)], [Test(4869215,aaaaa)])

sqlCtx.sql("select * FROM test").collectAsList() return [[], []]

任何帮助将不胜感激。我用 RDD 测试没有遇到任何问题我确实想从 RDD 迁移到 Dataframe，并且能够直接从 Spark 使用 Parquet 来存储文件

提前致谢

【问题讨论】：

标签： scala unit-testing apache-spark-sql

【解决方案1】：

测试通过，代码如下：

class TestDFSpec extends SharedSparkContext  {
  import sqlCtx.implicits._
  "Test DF " should {
    "pass equality" in {
      val createDF = sqlCtx.createDataFrame(Seq(create1,create2).map(Test.from))
      createDF.registerTempTable("test")
      val result = sqlCtx.sql("select * FROM test").collect()
      result === Array(Test.from(create1),Test.from(create2)).map(Row.fromTuple)
    }
  }

  val create1 = "4869215,bbbbb"
  val create2 = "4869215,aaaaa"
}

主要区别在于 DataFrame 的创建方式：从 Seq[Test] 而不是 RDD[Test]

我问了一些关于 spark 邮件的解释：http://apache-spark-user-list.1001560.n3.nabble.com/Unit-testing-dataframe-td24240.html#none

【讨论】：