RDD 용 DataFrame을 만들 수 없습니다.

Question

동적 스키마 생성을 사용하여 Dataframe을 만들려고합니다. 여기 코드는 다음과 같습니다

내가 런타임에 이러한 예외가 계속하지만

def mapMetricList(row: Row): Seq[Metric] = ??? 

val fields = Seq("Field1", "Field2") 

case class Metric(name: String, count: Long) 
def convertMetricList(df: DataFrame): DataFrame = { 
    val outputFields = df.schema.fieldNames.filter(f => fields.contains(f)) 

    val rdd = df.rdd.map(row => { 
    val schema = row.schema 
    val metrics = mapMetricList(row) 
    val s = outputFields.map(name => row.get(schema.fieldIndex(name))) 
    Row.fromSeq(s ++ Seq(metrics)) 
    }) 

    val nonMetricsSchema = outputFields.map(f => df.schema.apply(f)) 
    val metricField = StructField("total",ArrayType(ScalaReflection.schemaFor[Metric].dataType.asInstanceOf[StructType]),nullable=true) 
    val schema = StructType(nonMetricsSchema ++ Seq(metricField)) 
    schema.printTreeString() 
    val dff = spark.createDataFrame(rdd, schema) 
    dff 
} 

: 나는 불꽃 2.1.0을 사용하고

Caused by: java.lang.RuntimeException: Metric is not a valid external type for schema of struct<name:string,count:bigint> 
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.evalIfCondExpr3$(Unknown Source) 
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.evalIfFalseExpr4$(Unknown Source) 
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$SpecificUnsafeProjection.apply(Unknown Source) 
    at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder.toRow(ExpressionEncoder.scala:290) 

을

Answer 1

내 컴퓨터에 스파크 1.6 괜찮 았는데, I "convertMetricList"함수의 결과를 인쇄합니다. "metricField"필드의 "count"유형에서 문제가 발생할 수 있습니다. 언급 당신의 추적 "BIGINT"에서, 내 ENV 유형에 "LongType"입니다 :

이

StructField(total,ArrayType(
    StructType(StructField(name,StringType,true), 
    StructField(count,LongType,false) 
),true),true) 

당신은 당신의 ENV의 "metricField"유형을 확인할 수 있습니다. 다른 경우, 해결 방법은 Metric 구조를 하드 코딩하는 것입니다.