一部のRowオブジェクトに異なる数の要素が含まれている可能性がある場合、RDD からスパークデータフレームDataset を準備するスキーマをプログラムで推論します

spark rddでneo4jノード情報を取得していますneo4j-sparkコネクタ。 RDD<Row> を入手できます呼び出すことにより loadNodeRdds() 方法。しかし、データフレーム呼び出しを取得しようとすると loadDataframe() 方法、例外をスローします（主な質問が最終的に異なることが判明する可能性があるため、スタックトレースが長すぎる場合は、スタックトレースをスキップします）。

java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: java.util.Collections$UnmodifiableMap is not a valid external type for schema of string
if (assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object).isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, Condition), StringType), true) AS Condition#4
+- if (assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object).isNullAt) null else staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, Condition), StringType), true)
   :- assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object).isNullAt
   :  :- assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object)
   :  :  +- input[0, org.apache.spark.sql.Row, true]
   :  +- 0
   :- null
   +- staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, Condition), StringType), true)
      +- validateexternaltype(getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, Condition), StringType)
         +- getexternalrowfield(assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object), 0, Condition)
            +- assertnotnull(input[0, org.apache.spark.sql.Row, true], top level row object)
               +- input[0, org.apache.spark.sql.Row, true]
(skipped a lot of rows as it made question reach its character limit)

上記の大きなスタックレースから多くを得ることができませんでした。

だから私は JavaRDD<Row> を取りましたそしてそれを DataFrame<Row> に変換してみましたプログラムで指定することにより StructType スキーマ。

StructType schema = loadSchema();
Dataset<Row> df = ss.createDataFrame(neo4jJavaRdd , schema);

これはやや似た例外を投げました。

私がやったことは、単一のneo4jノードの個々のプロパティを取得し、 Row を準備したことですそして、 JavaRDD<Row> それから、次のようにプログラムでスキーマを指定して、そこからデータフレームを作成しようとしました：

Row row1 = RowFactory.create("val1", " val2", "val3", "val4", "val5", "val6", 152214d, "val7", 152206d, 11160d, "val8");
List<StructField> fields = new ArrayList<StructField>();
fields.add(DataTypes.createStructField("attr1", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("attr2", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("attr3", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("attr4", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("attr5", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("attr6", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("attrd1", DataTypes.DoubleType, true));
fields.add(DataTypes.createStructField("attr7", DataTypes.StringType, true));
fields.add(DataTypes.createStructField("attrd2", DataTypes.DoubleType, true));
fields.add(DataTypes.createStructField("attrd3", DataTypes.DoubleType, true));
fields.add(DataTypes.createStructField("attr8", DataTypes.StringType, true));

これはうまくいきました。

だから私はすべてのノードをチェックし、すべてのノードではないことに気付きました（つまり、すべて Row s in JavaRDD<Row> ）同じ数の属性があります。これにより、データフレームの準備が失敗する必要があります。必要なくプログラムで何らかの方法で処理できますかpojoを作成して指定する。