Spark Kryo-Encoder ArrayIndexOutOfBoundsException

Question

Ich versuche, ein Dataset mit einigen Geodaten mit Spark und ESRI zu erstellen. Wenn Foo nur Point Feld haben, wird es funktionieren, aber wenn ich einige andere Felder über eine Point hinaus hinzufügen, erhalte ich ArrayIndexOutOfBoundsException.

import com.esri.core.geometry.Point 
import org.apache.spark.sql.{Encoder, Encoders, SQLContext} 
import org.apache.spark.{SparkConf, SparkContext} 

object Main { 

    case class Foo(position: Point, name: String) 

    object MyEncoders { 
    implicit def PointEncoder: Encoder[Point] = Encoders.kryo[Point] 

    implicit def FooEncoder: Encoder[Foo] = Encoders.kryo[Foo] 
    } 

    def main(args: Array[String]): Unit = { 
    val sc = new SparkContext(new SparkConf().setAppName("app").setMaster("local")) 
    val sqlContext = new SQLContext(sc) 
    import MyEncoders.{FooEncoder, PointEncoder} 
    import sqlContext.implicits._ 
    Seq(new Foo(new Point(0, 0), "bar")).toDS.show 
    } 
} 

Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException: 1 bei org.apache.spark.sql.execution.Queryable $$ $ anonfun format $ 1 $ anonfun $$ $ 2.Apply gelten (Queryable.scala: 71) um org.apache.spark.sql.execution.Queryable $$ anonfun $ formatString $ 1 $$ anonfun $ gelten $ 2.apply (Queryable.scala: 70) um scala.collection.TraversableLike $ WithFilter $$ anonfun $ foreach $ 1.apply (TraversableLike.scala: 772) um scala.collection.mutable.ResizableArray $ class.foreach (ResizableArray.scal a: 59) bei scala.collection.mutable.ArrayBuffer.foreach (ArrayBuffer.scala: 47) bei scala.collection.TraversableLike $ WithFilter.foreach (TraversableLike.scala: 771) bei org.apache.spark. sql.execution.Queryable $$ anonfun $ formatString $ 1.apply (Queryable.scala: 70) um org.apache.spark.sql.execution.Queryable $$ anonfun $ formatString $ 1.apply (Queryable.scala: 69) at scala.collection.mutable.ArraySeq.foreach (ArraySeq.scala: 73) um org.apache.spark.sql.execution.Queryable $ class.formatString (Queryable.scala: 69) bei org.apache.spark.sql .Dataset.formatString (Dataset.scala: 65) um org.apache.spark.sql.Dataset.showString (Dataset.scala: 263) um org.apache.spark.sql.Dataset.show (Dataset.scala: 230) um org.apache.spark.sql.Dataset.show (Dataset.scala: 193) um org.apache.spark.sql.Dataset .show (Dataset.scala: 201) bei Haupt $ .main (Main.scala: 24) bei Main.main (Main.scala)

Answer 1

Kryo-Encoder für auf Spark SQL Data Types Basis komplexer Datentypen erstellen. So überprüfen Sie das Ergebnis des Schemas, das schaffen Kryo:

val enc: Encoder[Foo] = Encoders.kryo[Foo] 
println(enc.schema) // StructType(StructField(value,BinaryType,true)) 
val numCols = schema.fieldNames.length // 1 

So haben Sie eine Spalte Daten in Datensatz und es ist im binären Format. Aber es ist seltsam, dass Spark versucht, Dataset in mehr als einer Spalte anzuzeigen (und dieser Fehler auftritt). Um dies zu beheben, aktualisieren Sie die Spark-Version auf 2.0.0.

Mit Spark 2.0.0 haben Sie immer noch Probleme mit Spalten Datentypen. Ich hoffe, dass das Schreiben manuelle Schema funktioniert, wenn Sie StructType für esri Point-Klasse schreiben kann:

val schema = StructType(
    Seq(
    StructField("point", StructType(...), true), 
    StructField("name", StringType, true) 
    ) 
) 

val rdd = sc.parallelize(Seq(Row(new Point(0,0), "bar"))) 

sqlContext.createDataFrame(rdd, schema).toDS