从嵌套的用户定义对象中创建DataFrame

Question

我正在使用MongoDB-Hadoop连接器读取具有嵌入文档的集合。

{
    "user_id" : NumberLong(2),
    "persona_created" : true,
    "persona_createdAt" : ISODate("2016-02-24T06:41:49.761Z"),
    "persona" : [{"persona_type" : 1,
            "created_using_algo" : "Name of the algo",
            "version_algo" : "1.0",
            "createdAt" : ISODate("2016-02-24T06:41:49.761Z"),
             "persona_items": {"key1":"value1", "key2": "value2"} }]
}

 class Persona_Items implements Serializable
 {
   private int key1;
   private String key2;
   // Getter/Setter and constructor 
 }

 class Persona implements Serializable
 {
   String persona_type;
   String created_using_algo
   String version_algo
   long createdAt;
   List<Persona_Items> listPersonaItems; 
     // Getter/setter and constructor
 }

 class PersonaMetaData implements Serializable
 {
   long user_id;
   boolean persona_created;
   long persona_createdAt;
   List<Persona> listPersona;

   // Getter/setter and constructor 
 }

 // RDD representing the complete collection
 JavaPairRDD<Object, BSONObject> bsonRdd = sc.newAPIHadoopRDD(inputConfig,
                     com.mongodb.hadoop.MongoInputFormat.class,
                     Object.class, BSONObject.class);

 // Get RDD of PersonaMetaData
 JavaRDD<PersonaMetaData> metaDataSchemaJavaRDD =
 bsonRdd.map(new Function<Tuple2<Object, BSONObject>, PersonaMetaData >() {
                         @Override
                         public PersonaMetaData  call(Tuple2<Object, BSONObject> objectBSONObjectTuple2)
                                 throws Exception { // Parse the BSON object and return a new PersonaMetaData object }

 // Convert into DataFrame

 dataFrame= sqlContext.createDataFrame(metaDataSchemaJavaRDD,
                     PersonaMetaData.class);

例外

scala.matcherror:io.abc.spark.schema.personametadata@31FF5060（类为io.abc.spark.sql.catalyst.catalyst.catalysttypeConverters$structconverters.scala:255)在org.apache.spark.sql.catalyst.catalysttypeConverters$structconverters.scala:255)在org.apache.spark.sql.catalyst.catalysttypeConverters spark.sql.catalyst.catalysttypeConverters$arrayConverter.toCatalystimpl（catalysttypeConverters.scala:169)在org.apache.spark.sql.catalyst.catalysttypeConverters$arrayConverters.scala:169)在org.apache.spark.sql.catalyst.catalysttypeConverters$arrayConverters.toCatalystimpl（catalysttypeConverters在org.apache.spark.sql.sqlcontext$$anonfun$9$$anonfun$applic$1$$anonfun$applic$2..applic(sqlcontext.scala:500)在org.apache.spark.sql.sqlcontext$$anonfun$applic$9$$anonfun$applic$1$anonfun$applic$2.applic(sqlcontext.scala:500)在scala.collection.traversablelike$$anonfun$map$1.applic(sqlcontext.scala:244)在.scala:33)在scala.collection.mutable.arrayops$ofref.foreach(arrayops.scala:108)在scala.collection.traversablelike$class.map(traversablelike.scala:244)在scala.collection.mutable.arrayops$ofref.map(arrayops.scala:108)在

类中没有任何列表运行时不会出现任何问题。

Answer 1

正如在Spark SQL的使用反射推断模式一节中所明确指出的，数据规则和数据集指南（DataFrames and Datasets Guide）

Spark SQL不支持包含嵌套或复杂类型（如列表或数组）的JavaBeans。