scala - 如何在 Spark-Scala 中创建具有已定义架构的行？

Question

我想使用案例类中的模式创建一个行来测试我的 map 函数之一。我能想到的最直接的方法是:

import org.apache.spark.sql.Row

case class MyCaseClass(foo: String, bar: Option[String])

def buildRowWithSchema(record: MyCaseClass): Row = {
    sparkSession.createDataFrame(Seq(record)).collect.head
}

但是，仅获取单个行似乎开销很大，因此我研究了如何直接创建具有架构的行。这导致我:

import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.{Encoders, Row}

def buildRowWithSchemaV2(record: MyCaseClass): Row = {
    val recordValues: Array[Any] = record.getClass.getDeclaredFields.map((field) => {
        field.setAccessible(true)
        field.get(record)
    })
    new GenericRowWithSchema(recordValues, Encoders.product[MyCaseClass].schema)
}

不幸的是，第二个版本返回的 Row 与第一个 Row 不同。第一个版本中的选项字段被缩减为原始值，而第二个版本中它们仍然是选项。此外，第二个版本非常笨拙。

有更好的方法吗？

Answer 1

第二个版本为 bar 案例类字段返回 Option 本身，因此您不会像第一个版本那样获得原始值。您可以使用以下代码获取原始值

def buildRowWithSchemaV2(record: MyCaseClass): Row = {
  val recordValues: Array[Any] = record.getClass.getDeclaredFields.map((field) => {
    field.setAccessible(true)
    val returnValue = field.get(record)
    if(returnValue.isInstanceOf[Option[String]]){
      returnValue.asInstanceOf[Option[String]].get
    }
    else
      returnValue
  })
  new GenericRowWithSchema(recordValues, Encoders.product[MyCaseClass].schema)
}

但同时我会建议您使用 DataFrame 或 DataSet。

DataFrame 和 DataSet 本身就是 Row with schema 的集合。
所以当你定义了一个case class时，你只需要将你的输入数据编码到case class例如:假设您输入数据为

val data = Seq(("test1", "value1"),("test2", "value2"),("test3", "value3"),("test4", null))

如果您有一个文本文件，您可以根据需要使用 sparkContext.textFile 和 split 来阅读它。
现在，当您将数据转换为 RDD 时，将其转换为 dataframe 或 dataset 是两行代码

import sqlContext.implicits._
val dataFrame = data.map(d => MyCaseClass(d._1, Option(d._2))).toDF

.toDS 会生成dataset因此，您拥有 Rows with schema 的集合
为了进行验证，您可以执行以下操作

println(dataFrame.schema)    //for checking if there is schema

println(dataFrame.take(1).getClass.getName) //for checking if it is a collection of Rows

希望你有正确的答案。