如何在Spark 2.1中为集合编写编码器？

import org.apache.spark.sql.{Encoder, Encoders, SparkSession}

// 自定义类，表示数据集合的元素类型
case class Person(name: String, age: Int)

// 实现Encoder接口
class PersonEncoder extends Encoder[Person] {
  // 定义数据集合元素的结构
  def schema: org.apache.spark.sql.types.StructType = {
    Encoders.product[Person].schema
  }
  
  // 将数据集合元素转换为二进制格式
  def encode(person: Person): Array[Byte] = {
    Encoders.product[Person].serializer.apply(person).asInstanceOf[Array[Byte]]
  }
  
  // 从二进制格式中解码数据集合元素
  def decode(bytes: Array[Byte]): Person = {
    Encoders.product[Person].deserializer.apply(bytes)
  }
}

object Main {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .appName("EncoderExample")
      .master("local")
      .getOrCreate()
      
    import spark.implicits._
    
    // 创建数据集合
    val data = Seq(Person("Alice", 25), Person("Bob", 30), Person("Charlie", 35))
    
    // 使用自定义的编码器
    val encoder = new PersonEncoder()
    val encodedData = spark.createDataset(data)(encoder)
    
    // 打印编码后的数据集合
    encodedData.show()
    
    // 解码数据集合
    val decodedData = encodedData.map(encoder.decode)
    
    // 打印解码后的数据集合
    decodedData.show()
    
    spark.stop()
  }
}