序列化/反序列化 spark sql 数据帧的现有类

Question

使用火花 1.6.0 假设我有这样的课程

case class MyClass(date: java.util.Date, oid: org.bson.types.ObjectId)

如果我有

//rdd: RDD[MyClass]
rdd.toDF("date", "oid")

我收到java.lang.UnsupportedOperationException: Schema for type java.util.Date/org.bson.types.ObjectId is not supported

现在我知道我可以将其设为 java.sql.Date，但假设 MyClass 在太多其他地方被依赖以在任何地方进行更改，这仍然无法解决 ObjectId 问题。

我也知道UserDefinedType 选项。但这似乎只有在您还创建一个新类来使用它时才有效（同样，MyClass 的签名需要保持不变）

有没有办法只为java.util.Date 和org.bson.types.ObjectId 注册一个序列化器/反序列化器，这样我就可以在RDD[MyClass] 上调用toDF 并且它会正常工作？

UPDATE

所以这并不能完全回答我的问题，但它让我们畅通无阻，所以将在这里分享，希望对其他人有所帮助。所以大多数 json 库都支持这个用例，spark-sql 有一个内置的sqlContext.read.json(stringRdd).write.parquet("/path/to/output")。所以你可以使用你选择的 json 库为类定义（de）ser，序列化为字符串，然后 spark-sql 可以处理其余的

Answer 1

这取决于你所说的工作是什么意思。要序列化/反序列化一个对象，您只需要一个相应的UserDefinedType 和适当的注释。例如这样的：

@SQLUserDefinedType(udt = classOf[MyClassUDT])
case class MyClass(date: java.util.Date, oid: ObjectId)

class MyClassUDT extends UserDefinedType[MyClass] {
  override def sqlType: StructType = StructType(Seq(
    StructField("date", DateType, nullable = false),
    StructField("oid", StringType, nullable = false)
  ))

  override def serialize(obj: Any): InternalRow = {
    obj match {
      case MyClass(date, oid) =>
        val row = new GenericMutableRow(2)
        row(0) = new java.sql.Date(date.getTime)
        row(1) = UTF8String.fromString(oid.toString)
        row
    }
  }

  override def deserialize(datum: Any): MyClass = {
    datum match {
      case row: InternalRow =>
        val date: java.util.Date = new java.util.Date(
          row.get(0, DateType).asInstanceOf[java.sql.Date].getTime()
        )
        val oid = new ObjectId(row.getString(1))
        MyClass(date, oid)
    }
  }

  override def userClass: Class[MyClass] = classOf[MyClass]
}

这并不意味着您将能够访问定义在类或其任何字段上的任何方法。为此，您需要 UDF。

更接近无缝集成的是 Spark 数据集，但 AFAIK 尚无法定义自定义编码器。