根据介绍 Spark 数据集：
由于我们对 Spark 2.0 的期待，我们计划对 Datasets 进行一些激动人心的改进，特别是： ... 自定义编码器——虽然我们目前为多种类型自动生成编码器，但我们希望为自定义对象开放一个 API。
并尝试存储自定义类型Dataset导致以下错误，例如：
找不到存储在数据集中的类型的编码器。通过导入 sqlContext.implicits 支持原始类型（Int、String 等）和产品类型（案例类）。未来版本中将添加对序列化其他类型的支持
或者：
Java.lang.UnsupportedOperationException：没有找到用于 .... 的编码器
是否有任何现有的解决方法？
请注意，此问题仅作为社区 Wiki 答案的入口点存在。随时更新/改进问题和答案。
Set这个答案仍然有效且信息丰富，尽管自 2.2/2.3 以来情况有所好转，它增加了对、Seq、Map、Date、Timestamp和的内置编码器支持BigDecimal。如果你坚持只使用 case 类和通常的 Scala 类型来制作类型，那么你应该只使用隐式 in 就可以了SQLImplicits。
不幸的是，几乎没有添加任何东西来帮助解决这个问题。搜索或查找主要与原始类型有关的内容（以及对案例类的一些调整）@since 2.0.0。所以，首先要说的是：目前对自定义类编码器没有真正好的支持。有了这些，接下来是一些技巧，考虑到我们目前可以使用的东西，这些技巧可以做得尽可能好。作为一个预先的免责声明：这不会完美地工作，我会尽我最大的努力使所有限制都清楚和预先。Encoders.scalaSQLImplicits.scala
当你想创建一个数据集时，Spark“需要一个编码器（将类型 T 的 JVM 对象转换为内部 Spark SQL 表示），该编码器通常通过 a 的隐式自动创建SparkSession，或者可以通过调用静态方法显式创建on Encoders"（取自关于 的文档createDataset）。编码器将采用您正在编码的类型的Encoder[T]形式T。第一个建议是添加import spark.implicits._（它为您提供这些隐式编码器），第二个建议是使用这组编码器相关函数显式传入隐式编码器。
没有可用于常规课程的编码器，因此
import spark.implicits._
class MyObj(val i: Int)
// ...
val d = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))
会给你以下隐式相关的编译时错误：
找不到存储在数据集中的类型的编码器。通过导入 sqlContext.implicits 支持原始类型（Int、String 等）和产品类型（案例类）。未来版本中将添加对序列化其他类型的支持
但是，如果您将刚刚用于获取上述错误的任何类型包装在扩展的某个类中Product，则错误会令人困惑地延迟到运行时，因此
import spark.implicits._
case class Wrap[T](unwrap: T)
class MyObj(val i: Int)
// ...
val d = spark.createDataset(Seq(Wrap(new MyObj(1)),Wrap(new MyObj(2)),Wrap(new MyObj(3))))
编译得很好，但在运行时失败
java.lang.UnsupportedOperationException：未找到 MyObj 的编码器
原因是 Spark 使用隐式创建的编码器实际上仅在运行时生成（通过 scala relfection）。在这种情况下，Spark 在编译时的所有检查是最外层的类扩展Product（所有案例类都这样做），并且只在运行时意识到它仍然不知道该怎么做MyObj（如果我试图做同样的问题会发生a Dataset[(Int,MyObj)]- Spark 等到运行时才开始MyObj)。这些是急需解决的核心问题：
Product编译的类，尽管总是在运行时崩溃，并且MyObj以便它知道如何编码Wrap[MyObj]or (Int,MyObj)）。kryo每个人都建议的解决方案是使用kryo编码器。
import spark.implicits._
class MyObj(val i: Int)
implicit val myObjEncoder = org.apache.spark.sql.Encoders.kryo[MyObj]
// ...
val d = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))
不过，这很快就会变得非常乏味。特别是如果您的代码正在操作各种数据集、连接、分组等。您最终会积累一堆额外的隐式。那么，为什么不直接隐式地自动完成这一切呢？
import scala.reflect.ClassTag
implicit def kryoEncoder[A](implicit ct: ClassTag[A]) =
org.apache.spark.sql.Encoders.kryo[A](ct)
现在，似乎我几乎可以做任何我想做的事情（下面的例子在自动导入的spark-shell地方不起作用）spark.implicits._
class MyObj(val i: Int)
val d1 = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))
val d2 = d1.map(d => (d.i+1,d)).alias("d2") // mapping works fine and ..
val d3 = d1.map(d => (d.i, d)).alias("d3") // .. deals with the new type
val d4 = d2.joinWith(d3, $"d2._1" === $"d3._1") // Boom!
或者差不多。问题是使用kryo导致 Spark 只是将数据集中的每一行存储为一个平面二进制对象。对于map, filter，foreach这就足够了，但是对于像join，Spark 之类的操作确实需要将它们分成列。检查d2or的架构d3，您会看到只有一个二进制列：
d2.printSchema
// root
// |-- value: binary (nullable = true)
因此，使用 Scala 中隐含的魔力（更多内容在6.26.3 重载解决方案中），我可以为自己制作一系列隐含，这些隐含将尽可能好地完成工作，至少对于元组而言，并且可以很好地与现有的隐含一起工作：
import org.apache.spark.sql.{Encoder,Encoders}
import scala.reflect.ClassTag
import spark.implicits._ // we can still take advantage of all the old implicits
implicit def single[A](implicit c: ClassTag[A]): Encoder[A] = Encoders.kryo[A](c)
implicit def tuple2[A1, A2](
implicit e1: Encoder[A1],
e2: Encoder[A2]
): Encoder[(A1,A2)] = Encoders.tuple[A1,A2](e1, e2)
implicit def tuple3[A1, A2, A3](
implicit e1: Encoder[A1],
e2: Encoder[A2],
e3: Encoder[A3]
): Encoder[(A1,A2,A3)] = Encoders.tuple[A1,A2,A3](e1, e2, e3)
// ... you can keep making these
然后，有了这些隐式，我可以让我上面的例子工作，尽管有一些列重命名
class MyObj(val i: Int)
val d1 = spark.createDataset(Seq(new MyObj(1),new MyObj(2),new MyObj(3)))
val d2 = d1.map(d => (d.i+1,d)).toDF("_1","_2").as[(Int,MyObj)].alias("d2")
val d3 = d1.map(d => (d.i ,d)).toDF("_1","_2").as[(Int,MyObj)].alias("d3")
val d4 = d2.joinWith(d3, $"d2._1" === $"d3._1")
我还没有弄清楚如何在不重命名的情况下默认获取预期的元组名称 ( _1, _2, ...) - 如果其他人想玩这个，这是引入名称"value"的地方，也是元组的地方通常会添加名称。然而，关键是我现在有一个很好的结构化模式：
d4.printSchema
// root
// |-- _1: struct (nullable = false)
// | |-- _1: integer (nullable = true)
// | |-- _2: binary (nullable = true)
// |-- _2: struct (nullable = false)
// | |-- _1: integer (nullable = true)
// | |-- _2: binary (nullable = true)
因此，总而言之，此解决方法：
kryo）import spark.implicits._（涉及一些重命名）kyro那些可能有的字段了.toDF、指定新列名和转换回数据集来撤消此操作 - 并且模式名称似乎通过连接保留，最需要它们的地方）。这个不太愉快，也没有好的解决方案。但是，既然我们有了上面的元组解决方案，我预感来自另一个答案的隐式转换解决方案也不会那么痛苦，因为您可以将更复杂的类转换为元组。然后，在创建数据集之后，您可能会使用数据框方法重命名列。如果一切顺利，这确实是一个进步，因为我现在可以在我的类的字段上执行连接。如果我只使用一个平面二进制kryo序列化器，那是不可能的。
这是一个做所有事情的例子：我有一个类MyObj，它的字段类型Int为 、java.util.UUID和Set[String]。第一个照顾自己。第二个，虽然我可以序列化 usingkryo如果存储为 a 会更有用String（因为UUIDs 通常是我想要加入的东西）。第三个真的只是属于二进制列。
class MyObj(val i: Int, val u: java.util.UUID, val s: Set[String])
// alias for the type to convert to and from
type MyObjEncoded = (Int, String, Set[String])
// implicit conversions
implicit def toEncoded(o: MyObj): MyObjEncoded = (o.i, o.u.toString, o.s)
implicit def fromEncoded(e: MyObjEncoded): MyObj =
new MyObj(e._1, java.util.UUID.fromString(e._2), e._3)
现在，我可以使用这种机器创建一个具有良好模式的数据集：
val d = spark.createDataset(Seq[MyObjEncoded](
new MyObj(1, java.util.UUID.randomUUID, Set("foo")),
new MyObj(2, java.util.UUID.randomUUID, Set("bar"))
)).toDF("i","u","s").as[MyObjEncoded]
架构向我展示了具有正确名称的列和前两个我可以加入的东西。
d.printSchema
// root
// |-- i: integer (nullable = false)
// |-- u: string (nullable = true)
// |-- s: binary (nullable = true)
使用通用编码器。
目前有两种通用编码器可用，kryo后javaSerialization一种被明确描述为：
效率极低，只能作为最后的手段。
假设以下课程
class Bar(i: Int) {
override def toString = s"bar $i"
def bar = i
}
您可以通过添加隐式编码器来使用这些编码器：
object BarEncoders {
implicit def barEncoder: org.apache.spark.sql.Encoder[Bar] =
org.apache.spark.sql.Encoders.kryo[Bar]
}
可以按如下方式一起使用：
object Main {
def main(args: Array[String]) {
val sc = new SparkContext("local", "test", new SparkConf())
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
import BarEncoders._
val ds = Seq(new Bar(1)).toDS
ds.show
sc.stop()
}
}
它将对象存储为binary列，因此当转换为DataFrame您获得以下架构时：
root
|-- value: binary (nullable = true)
也可以使用kryo特定字段的编码器对元组进行编码：
val longBarEncoder = Encoders.tuple(Encoders.scalaLong, Encoders.kryo[Bar])
spark.createDataset(Seq((1L, new Bar(1))))(longBarEncoder)
// org.apache.spark.sql.Dataset[(Long, Bar)] = [_1: bigint, _2: binary]
请注意，我们在这里不依赖隐式编码器，而是显式传递编码器，因此这很可能不适用于toDS方法。
使用隐式转换：
提供可以编码的表示和自定义类之间的隐式转换，例如：
object BarConversions {
implicit def toInt(bar: Bar): Int = bar.bar
implicit def toBar(i: Int): Bar = new Bar(i)
}
object Main {
def main(args: Array[String]) {
val sc = new SparkContext("local", "test", new SparkConf())
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
import BarConversions._
type EncodedBar = Int
val bars: RDD[EncodedBar] = sc.parallelize(Seq(new Bar(1)))
val barsDS = bars.toDS
barsDS.show
barsDS.map(_.bar).show
sc.stop()
}
}
相关问题：
您可以使用 UDTRegistration，然后使用案例类、元组等……所有这些都可以与您的用户定义类型一起正常工作！
假设您想使用自定义枚举：
trait CustomEnum { def value:String }
case object Foo extends CustomEnum { val value = "F" }
case object Bar extends CustomEnum { val value = "B" }
object CustomEnum {
def fromString(str:String) = Seq(Foo, Bar).find(_.value == str).get
}
像这样注册它：
// First define a UDT class for it:
class CustomEnumUDT extends UserDefinedType[CustomEnum] {
override def sqlType: DataType = org.apache.spark.sql.types.StringType
override def serialize(obj: CustomEnum): Any = org.apache.spark.unsafe.types.UTF8String.fromString(obj.value)
// Note that this will be a UTF8String type
override def deserialize(datum: Any): CustomEnum = CustomEnum.fromString(datum.toString)
override def userClass: Class[CustomEnum] = classOf[CustomEnum]
}
// Then Register the UDT Class!
// NOTE: you have to put this file into the org.apache.spark package!
UDTRegistration.register(classOf[CustomEnum].getName, classOf[CustomEnumUDT].getName)
然后使用它！
case class UsingCustomEnum(id:Int, en:CustomEnum)
val seq = Seq(
UsingCustomEnum(1, Foo),
UsingCustomEnum(2, Bar),
UsingCustomEnum(3, Foo)
).toDS()
seq.filter(_.en == Foo).show()
println(seq.collect())
假设您要使用多态记录：
trait CustomPoly
case class FooPoly(id:Int) extends CustomPoly
case class BarPoly(value:String, secondValue:Long) extends CustomPoly
...并像这样使用它：
case class UsingPoly(id:Int, poly:CustomPoly)
Seq(
UsingPoly(1, new FooPoly(1)),
UsingPoly(2, new BarPoly("Blah", 123)),
UsingPoly(3, new FooPoly(1))
).toDS
polySeq.filter(_.poly match {
case FooPoly(value) => value == 1
case _ => false
}).show()
您可以编写一个自定义 UDT，将所有内容编码为字节（我在这里使用 java 序列化，但检测 Spark 的 Kryo 上下文可能更好）。
首先定义UDT类：
class CustomPolyUDT extends UserDefinedType[CustomPoly] {
val kryo = new Kryo()
override def sqlType: DataType = org.apache.spark.sql.types.BinaryType
override def serialize(obj: CustomPoly): Any = {
val bos = new ByteArrayOutputStream()
val oos = new ObjectOutputStream(bos)
oos.writeObject(obj)
bos.toByteArray
}
override def deserialize(datum: Any): CustomPoly = {
val bis = new ByteArrayInputStream(datum.asInstanceOf[Array[Byte]])
val ois = new ObjectInputStream(bis)
val obj = ois.readObject()
obj.asInstanceOf[CustomPoly]
}
override def userClass: Class[CustomPoly] = classOf[CustomPoly]
}
然后注册它：
// NOTE: The file you do this in has to be inside of the org.apache.spark package!
UDTRegistration.register(classOf[CustomPoly].getName, classOf[CustomPolyUDT].getName)
然后就可以使用了！
// As shown above:
case class UsingPoly(id:Int, poly:CustomPoly)
Seq(
UsingPoly(1, new FooPoly(1)),
UsingPoly(2, new BarPoly("Blah", 123)),
UsingPoly(3, new FooPoly(1))
).toDS
polySeq.filter(_.poly match {
case FooPoly(value) => value == 1
case _ => false
}).show()
在 Java Bean 类的情况下，这可能很有用
import spark.sqlContext.implicits._
import org.apache.spark.sql.Encoders
implicit val encoder = Encoders.bean[MyClasss](classOf[MyClass])
现在您可以简单地将 dataFrame 读取为自定义 DataFrame
dataFrame.as[MyClass]
这将创建一个自定义类编码器，而不是二进制编码器。
编码器在Spark2.0. 并且Kryo仍然是推荐的serialization选择。
您可以使用 spark-shell 查看以下示例
scala> import spark.implicits._
import spark.implicits._
scala> import org.apache.spark.sql.Encoders
import org.apache.spark.sql.Encoders
scala> case class NormalPerson(name: String, age: Int) {
| def aboutMe = s"I am ${name}. I am ${age} years old."
| }
defined class NormalPerson
scala> case class ReversePerson(name: Int, age: String) {
| def aboutMe = s"I am ${name}. I am ${age} years old."
| }
defined class ReversePerson
scala> val normalPersons = Seq(
| NormalPerson("Superman", 25),
| NormalPerson("Spiderman", 17),
| NormalPerson("Ironman", 29)
| )
normalPersons: Seq[NormalPerson] = List(NormalPerson(Superman,25), NormalPerson(Spiderman,17), NormalPerson(Ironman,29))
scala> val ds1 = sc.parallelize(normalPersons).toDS
ds1: org.apache.spark.sql.Dataset[NormalPerson] = [name: string, age: int]
scala> val ds2 = ds1.map(np => ReversePerson(np.age, np.name))
ds2: org.apache.spark.sql.Dataset[ReversePerson] = [name: int, age: string]
scala> ds1.show()
+---------+---+
| name|age|
+---------+---+
| Superman| 25|
|Spiderman| 17|
| Ironman| 29|
+---------+---+
scala> ds2.show()
+----+---------+
|name| age|
+----+---------+
| 25| Superman|
| 17|Spiderman|
| 29| Ironman|
+----+---------+
scala> ds1.foreach(p => println(p.aboutMe))
I am Ironman. I am 29 years old.
I am Superman. I am 25 years old.
I am Spiderman. I am 17 years old.
scala> val ds2 = ds1.map(np => ReversePerson(np.age, np.name))
ds2: org.apache.spark.sql.Dataset[ReversePerson] = [name: int, age: string]
scala> ds2.foreach(p => println(p.aboutMe))
I am 17. I am Spiderman years old.
I am 25. I am Superman years old.
I am 29. I am Ironman years old.
直到现在]appropriate encoders目前范围内没有，所以我们的人没有被编码为binary值。但是一旦我们提供一些使用序列化的implicit编码器，这种情况就会改变。Kryo
// Provide Encoders
scala> implicit val normalPersonKryoEncoder = Encoders.kryo[NormalPerson]
normalPersonKryoEncoder: org.apache.spark.sql.Encoder[NormalPerson] = class[value[0]: binary]
scala> implicit val reversePersonKryoEncoder = Encoders.kryo[ReversePerson]
reversePersonKryoEncoder: org.apache.spark.sql.Encoder[ReversePerson] = class[value[0]: binary]
// Ecoders will be used since they are now present in Scope
scala> val ds3 = sc.parallelize(normalPersons).toDS
ds3: org.apache.spark.sql.Dataset[NormalPerson] = [value: binary]
scala> val ds4 = ds3.map(np => ReversePerson(np.age, np.name))
ds4: org.apache.spark.sql.Dataset[ReversePerson] = [value: binary]
// now all our persons show up as binary values
scala> ds3.show()
+--------------------+
| value|
+--------------------+
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
+--------------------+
scala> ds4.show()
+--------------------+
| value|
+--------------------+
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
|[01 00 24 6C 69 6...|
+--------------------+
// Our instances still work as expected
scala> ds3.foreach(p => println(p.aboutMe))
I am Ironman. I am 29 years old.
I am Spiderman. I am 17 years old.
I am Superman. I am 25 years old.
scala> ds4.foreach(p => println(p.aboutMe))
I am 25. I am Superman years old.
I am 29. I am Ironman years old.
I am 17. I am Spiderman years old.
我的示例将使用 Java，但我认为适应 Scala 并不困难。
只要是一个简单的Java Bean ，我就已经非常成功地转换RDD<Fruit>为Dataset<Fruit>使用spark.createDataset和Encoders.bean。Fruit
第 1 步：创建简单的 Java Bean。
public class Fruit implements Serializable {
private String name = "default-fruit";
private String color = "default-color";
// AllArgsConstructor
public Fruit(String name, String color) {
this.name = name;
this.color = color;
}
// NoArgsConstructor
public Fruit() {
this("default-fruit", "default-color");
}
// ...create getters and setters for above fields
// you figure it out
}
在 DataBricks 的人加强他们的编码器之前，我会坚持使用原始类型和字符串作为字段的类。如果您有一个带有嵌套对象的类，请创建另一个简单的 Java Bean，并将其所有字段都展平，这样您就可以使用 RDD 转换将复杂类型映射到更简单的类型。当然，这是一些额外的工作，但我想这将对使用平面模式的性能有很大帮助。
第 2 步：从 RDD 获取数据集
SparkSession spark = SparkSession.builder().getOrCreate();
JavaSparkContext jsc = new JavaSparkContext();
List<Fruit> fruitList = ImmutableList.of(
new Fruit("apple", "red"),
new Fruit("orange", "orange"),
new Fruit("grape", "purple"));
JavaRDD<Fruit> fruitJavaRDD = jsc.parallelize(fruitList);
RDD<Fruit> fruitRDD = fruitJavaRDD.rdd();
Encoder<Fruit> fruitBean = Encoders.bean(Fruit.class);
Dataset<Fruit> fruitDataset = spark.createDataset(rdd, bean);
瞧！起泡，冲洗，重复。
对于那些可能在我的情况下的人，我也将我的答案放在这里。
再具体一点，
我正在从 SQLContext 中读取“设置类型数据”。所以原始数据格式是DataFrame。
val sample = spark.sqlContext.sql("select 1 as a, collect_set(1) as b limit 1")
sample.show()
+---+---+
| a| b|
+---+---+
| 1|[1]|
+---+---+
然后使用带有 mutable.WrappedArray 类型的 rdd.map() 将其转换为 RDD。
sample
.rdd.map(r =>
(r.getInt(0), r.getAs[mutable.WrappedArray[Int]](1).toSet))
.collect()
.foreach(println)
结果：
(1,Set(1))
@Alec 的回答很棒！只是在他/她的答案的这一部分添加评论：
import spark.implicits._
case class Wrap[T](unwrap: T)
class MyObj(val i: Int)
// ...
val d = spark.createDataset(Seq(Wrap(new MyObj(1)),Wrap(new MyObj(2)),Wrap(new MyObj(3))))
@Alec 提到：
无法为嵌套类型传入自定义编码器（我无法为 Spark 提供仅用于 MyObj 的编码器，以便它知道如何编码 Wrap[MyObj] 或 (Int,MyObj)）。
似乎是这样，因为如果我为以下内容添加编码器MyObj：
implicit val myEncoder = org.apache.spark.sql.Encoders.kryo[MyObj]
，它仍然失败：
java.lang.UnsupportedOperationException: No Encoder found for MyObj
- field (class: "MyObj", name: "unwrap")
- root class: "Wrap"
at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:643)
但请注意重要的错误消息：
根类：“包装”
它实际上暗示了编码MyObj是不够的，您必须对整个链进行编码，包括Wrap[T].
所以如果我这样做，它可以解决问题：
implicit val myWrapperEncoder = org.apache.spark.sql.Encoders.kryo[Wrap[MyObj]]
因此，@Alec 的评论并非如此：
我无法为 Spark 提供仅用于 MyObj 的编码器，以便它知道如何编码 Wrap[MyObj] 或 (Int,MyObj)
我们仍然有办法为 Spark 提供编码器，MyObj以便它知道如何编码 Wrap[MyObj] 或 (Int,MyObj)。
除了已经给出的建议之外，我最近发现的另一个选择是您可以声明包含 trait 的自定义类org.apache.spark.sql.catalyst.DefinedByConstructorParams。
如果类的构造函数使用 ExpressionEncoder 可以理解的类型，即原始值和标准集合，则此方法有效。当您无法将类声明为案例类，但又不想在每次将其包含在数据集中时都使用 Kryo 对其进行编码时，它会派上用场。
例如，我想声明一个包含 Breeze 向量的案例类。唯一能够处理这种情况的编码器通常是 Kryo。但是，如果我声明了一个扩展 Breeze DenseVector 和 DefinedByConstructorParams 的子类，ExpressionEncoder 就会理解它可以被序列化为一个 Doubles 数组。
这是我声明它的方式：
class SerializableDenseVector(values: Array[Double]) extends breeze.linalg.DenseVector[Double](values) with DefinedByConstructorParams
implicit def BreezeVectorToSerializable(bv: breeze.linalg.DenseVector[Double]): SerializableDenseVector = bv.asInstanceOf[SerializableDenseVector]
现在，我可以使用SerializableDenseVector简单的 ExpressionEncoder 而不使用 Kryo 在数据集中（直接或作为产品的一部分）使用。它就像 Breeze DenseVector 一样工作，但序列化为 Array[Double]。