Spark kryo 编码器 ArrayIndexOutOfBoundsException

Question

我正在尝试使用 spark 和 esri 创建包含一些地理数据的数据集。如果Foo 只有Point 字段，它会起作用，但如果我在Point 之外添加一些其他字段，我会得到 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
  }
}

线程“main”中的异常 java.lang.ArrayIndexOutOfBoundsException: 1 在 org.apache.spark.sql.execution.Queryable$$anonfun$formatString$1$$anonfun$apply$2.apply(Queryable.scala:71) 在 org.apache.spark.sql.execution.Queryable$$anonfun$formatString$1$$anonfun$apply$2.apply(Queryable.scala:70) 在 scala.collection.TraversableLike$WithFilter$$anonfun$foreach$1.apply(TraversableLike.scala:772) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) 在 scala.collection.TraversableLike$WithFilter.foreach(TraversableLike.scala:771) 在 org.apache.spark.sql.execution.Queryable$$anonfun$formatString$1.apply(Queryable.scala:70) 在 org.apache.spark.sql.execution.Queryable$$anonfun$formatString$1.apply(Queryable.scala:69) 在 scala.collection.mutable.ArraySeq.foreach(ArraySeq.scala:73) 在 org.apache.spark.sql.execution.Queryable$class.formatString(Queryable.scala:69) 在 org.apache.spark.sql.Dataset.formatString(Dataset.scala:65) 在 org.apache.spark.sql.Dataset.showString(Dataset.scala:263) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:230) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:193) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:201) 在 Main$.main(Main.scala:24) 在 Main.main(Main.scala)

Answer 1

Kryo 基于Spark SQL Data Types 为复杂数据类型创建编码器。所以检查 kryo 创建的模式的结果：

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

所以你在数据集中有一列数据，它是二进制格式。但奇怪的是，为什么 Spark 试图在不止一列中显示数据集（并且发生了该错误）。要解决此问题，请将 Spark 版本升级到 2.0.0。

使用 Spark 2.0.0，列数据类型仍然存在问题。如果您可以为 esri Point 类编写 StructType，我希望编写手动模式有效：

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