对于Pojo的/原语，Spark数据集从Row的Encoder移动到Encoder。Catalyst引擎使用ExpressionEncoder转换SQL表达式中的列。然而，似乎没有其他Encoder子类可用作我们自己实现的模板。

以下是在Spark 1. X/DataFrames中满意的代码示例，该代码在新制度中无法编译：

//mapping each row to RDD tuple
df.map(row => {
    var id: String = if (!has_id) "" else row.getAs[String]("id")
    var label: String = row.getAs[String]("label")
    val channels  : Int = if (!has_channels) 0 else row.getAs[Int]("channels")
    val height  : Int = if (!has_height) 0 else row.getAs[Int]("height")
    val width : Int = if (!has_width) 0 else row.getAs[Int]("width")
    val data : Array[Byte] = row.getAs[Any]("data") match {
      case str: String => str.getBytes
      case arr: Array[Byte@unchecked] => arr
      case _ => {
        log.error("Unsupport value type")
        null
      }
    }
    (id, label, channels, height, width, data)
  }).persist(StorageLevel.DISK_ONLY)

}

我们得到一个编译器错误

Error:(56, 11) Unable to find encoder for type stored in a Dataset.
Primitive types (Int, String, etc) and Product types (case classes) are supported 
by importing spark.implicits._  Support for serializing other types will be added in future releases.
    df.map(row => {
          ^

所以不管怎样/某个地方应该有办法

• 定义/实现我们的自定义编码器
• 在DataFrame（现在是Row类型的数据集）上执行映射时应用它
• 注册编码器以供其他自定义代码使用

我正在寻找成功执行这些步骤的代码。