Spark : SQL 컨텍스트 : Scala 객체에서 데이터 프레임 만들기

Question

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. 
} 

나는 경우 클래스에 사람을 변환 주소 대신이라면 스파크는 위의 구문을 사용하거나 sc.parallelize(persons, 2).toDF를 사용하거나 내가 할 수 sqlContext.createDataFrame(sc.parallelize(persons, 2),StructType)

를 사용하여 자동으로을 스키마를 상속 할 수 있습니다 20 개 이상의 필드를 가질 수 없기 때문에 케이스 클래스를 사용하지 않으며 클래스에 많은 필드가 있습니다. 그리고 StructType을 사용하면 많은 불편을 겪습니다. 사례 클래스가 가장 편리하지만 속성을 너무 많이 보유 할 수는 없습니다.

미리 감사드립니다.

Answer 1

두 가지 코드를 변경하면 printSchema()가 대소 문자 클래스를 사용하지 않고도 데이터 프레임의 전체 구조를 방출합니다. 당신이 .toDF를 사용하여 dataframe을 만들어야합니다, 둘째

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 
    } 
} 

: 다니엘 제안

첫째로, 당신은 당신의 수업을 위해 scala.Product 특성을 확장해야합니다 (고통스러운, 그러나 아래 .toDF 방법에 필요) 오히려과 같이 sqlContext.createDataFrame(..)를 사용하는 것보다 import sqlContext.implicits._ 범위로하게된다 내재적 방법은

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

다음 claimDF.printSchema()을 출력한다 :

01,235,

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 개의 필드 제한을 제거 할 수 있습니다.

Answer 2

입력 해 주셔서 감사합니다.

더 큰 케이스 클래스를 지원하는 Scala 2.11을 사용하여 궁극적으로 Spark 2.1로 마이그레이션하여이 문제를 해결했습니다.

Spark 1.6 및 Scala 2.10의 경우 Dataframe을 구축하기 위해 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)) 
    ) 
}