Spark：SQL 上下文：从 Scala 对象创建数据框答案

【问题标题】：Spark : SQL Context : Creating a data frame from Scala objectSpark：SQL 上下文：从 Scala 对象创建数据框
【发布时间】：2017-06-09 07:07:54
【问题描述】：

我正在使用 Spark 1.5.2 使用以下语法从 scala 对象创建数据框。我的目的是为单元测试创建数据。

class Address (first:String = null, second: String = null, zip: String = null){}
class Person (id: String = null, name: String = null, address: Seq[Address] = null){}

def test () = {

  val sqlContext = new SQLContext(sc)

  import sqlContext.implicits._

  val persons = Seq(
    new Person(id = "1", name = "Salim", 
      address = Seq(new Address(first = "1st street"))),
    new Person(name = "Sana",
      address = Seq(new Address(zip = "60088")))
  )

  // The code can't infer schema automatically
  val claimDF = sqlContext.createDataFrame(sc.parallelize(persons, 2),classOf[Person])

  claimDF.printSchema() // This prints "root" not the schema of Person.
}

如果我将 Person 和 Address 转换为 case 类，则 Spark 可以使用上述语法或使用 sc.parallelize(persons, 2).toDF 或使用 sqlContext.createDataFrame(sc.parallelize(persons, 2),StructType) 自动继承架构

我不能使用案例类，因为它不能容纳超过 20 个字段，而且我在类中有很多字段。而且使用 StructType 会带来很多不便。案例类最方便，但不能容纳太多属性。

请帮忙，提前谢谢。

【问题讨论】：

我认为，如果您的类扩展 Product trait 并实现其抽象方法，它可能会起作用。（因为这个签名：createDataFrame[A <: Product](data: Seq[A])）

标签： apache-spark spark-dataframe

【解决方案1】：

非常感谢您的意见。

我们最终迁移到带有 Scala 2.11 的 Spark 2.1，它支持更大的案例类，因此这个问题得到了解决。

对于 Spark 1.6 和 Scala 2.10，我最终构建了 Row 对象和 Struct 类型来构建数据框。

val rows = Seq(Row("data"))
val aRDD = sc.parallelize(rows)
val aDF = sqlContext.createDataFrame(aRDD,getSchema())

def getSchema(): StructType= {
    StructType(
        Array(
            StructField("jobNumber", StringType, nullable = true))
    )
}

【讨论】：

【解决方案2】：

对您的代码进行两项更改将使 printSchema() 在不使用案例类的情况下发出数据帧的完整结构。

首先，正如 Daniel 所建议的，您需要让您的类扩展 scala.Product 特征（很痛苦，但对于下面的 .toDF 方法是必需的）：

class Address (first:String = null, second: String = null, zip: String = null) extends Product with Serializable
{
  override def canEqual(that: Any) = that.isInstanceOf[Address]
  override def productArity: Int = 3
  def productElement(n: Int) = n match {
    case 0 => first; case 1 => second; case 2 => zip
  }
}

class Person (id: String = null, name: String = null, address: Seq[Address] = null) extends Product with Serializable
{
  override def canEqual(that: Any) = that.isInstanceOf[Person]
  override def productArity: Int = 3
  def productElement(n: Int) = n match {
    case 0 => id; case 1 => name; case 2 => address
  }
}

其次，您应该使用 .toDF 隐式方法创建数据框，该方法通过 import sqlContext.implicits._ 引入范围，而不是像这样使用 sqlContext.createDataFrame(..)：

val claimDF = sc.parallelize(persons, 2).toDF

然后 claimDF.printSchema() 将打印：

root
 |-- id: string (nullable = true)
 |-- name: string (nullable = true)
 |-- address: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- first: string (nullable = true)
 |    |    |-- second: string (nullable = true)
 |    |    |-- zip: string (nullable = true)

或者，您可以使用 Scala 2.11.0-M3，它取消了案例类的 22 个字段限制。

【讨论】：