스파크 : 행의 열 중 하나가 각 행에 대해 다음과 같은 데이터 목록

Question

나는 RDD [행]을 가지고있다 dataframe에 RDD [행] 변환

[guid, List(peopleObjects)] 
["123", List(peopleObjects1, peopleObjects2, peopleObjects3)] 

나는이를 변환 할

dataframe
내 스키마 대신 StringType에 대해 서로 다른 데이터 유형을 사용하여 I이어야한다 다음 코드

val personStructureType = new StructType() 
    .add(StructField("guid", StringType, true)) 
    .add(StructField("personList", StringType, true)) 
val personDF = hiveContext.createDataFrame(personRDD, personStructureType) 

을 사용하고?

내 목록이 작동 단지 문자열 인 경우

하지만이 목록 나는 다음과 같은 오류 그것은 당신이 뭘 하려는지 전혀 분명하지 않다

scala.MatchError: List(personObject1, personObject2, personObject3) (of class scala.collection.immutable.$colon$colon) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StringConverter$.toCatalystImpl(CatalystTypeConverters.scala:295) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StringConverter$.toCatalystImpl(CatalystTypeConverters.scala:294) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$CatalystTypeConverter.toCatalyst(CatalystTypeConverters.scala:102) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:260) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StructConverter.toCatalystImpl(CatalystTypeConverters.scala:250) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$CatalystTypeConverter.toCatalyst(CatalystTypeConverters.scala:102) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$$anonfun$createToCatalystConverter$2.apply(CatalystTypeConverters.scala:401) 
    at org.apache.spark.sql.SQLContext$$anonfun$7.apply(SQLContext.scala:445) 
    at org.apache.spark.sql.SQLContext$$anonfun$7.apply(SQLContext.scala:445) 
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328) 
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328) 
    at scala.collection.Iterator$$anon$11.next(Iterator.scala:328) 
    at org.apache.spark.util.collection.ExternalSorter.insertAll(ExternalSorter.scala:219) 
    at org.apache.spark.shuffle.sort.SortShuffleWriter.write(SortShuffleWriter.scala:73) 
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:73) 
    at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:41) 
    at org.apache.spark.scheduler.Task.run(Task.scala:88) 
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
    at java.lang.Thread.run(Thread.java:745) 

Answer 1

하지만 당신이 무엇을 할 수있는 더 좋은 방법을 얻을 때 그러면 case class을 만들고 RDD 행을 case class에 매핑 한 다음 toDF으로 전화하면됩니다. 같은

뭔가 :

case class MyClass(guid: Int, peopleObjects: List[String]) 

val rdd = sc.parallelize(Array((123,List("a","b")),(1232,List("b","d")))) 

val df = rdd.map(r => MyClass(r._1, r._2)).toDF 
df.show 
+----+-------------+ 
|guid|peopleObjects| 
+----+-------------+ 
| 123|  [a, b]| 
|1232|  [b, d]| 
+----+-------------+ 

또는 당신은 그것을 오랫동안 손 방법을 할 수 있지만이 같은 경우 클래스를 사용하지 않고 :

val df = sqlContext.createDataFrame(
    rdd.map(r => Row(r._1, r._2)), 
    StructType(Array(
    StructField("guid",IntegerType), 
    StructField("peopleObjects", ArrayType(StringType)) 
)) 
)