Spark Java编码器。bean无法转换为Scala定义的类

Question

我有Java代码将JavaRDD转换为Dataset并将其保存到HDFS：

Dataset<User> userDataset = sqlContext.createDataset(userRdd.rdd(), Encoders.bean(User.class));
userDataset.write.json("some_path");

User类是用Scala语言定义的：

case class User(val name: Name, val address: Seq[Address]) extends Serializable

case class Name(firstName: String, lastName: Option[String])

case class Address(address: String)

代码符合并成功运行，文件保存到HDFS，而输出文件中的User类具有空架构：

val users = spark.read.json("some_path")
users.count // 100,000 which is same as "userRdd"
users.printSchema // users: org.apache.spark.sql.DataFrame = []

为什么Encoders.bean在这种情况下不起作用

Answer 1

Encoders.bean不支持Scala case类，Encoders.product支持该类Encoders.product将TypeTag作为参数，而初始化TypeTag在Java中是不可能的。我创建了一个Scala对象来提供TypeTag：

import scala.reflect.runtime.universe._

object MyTypeTags {
  val UserTypeTag: TypeTag[User] = typeTag[User]
}

然后在Java代码中：Dataset<User> userDataset = sqlContext.createDataset(userRdd.rdd(), Encoders.product(MyTypeTags.UserTypeTag()));