如何为Spark Dataframe创建自定义编写器？

import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql.Encoder
import org.apache.spark.sql.catalyst.expressions.{Expression, GenericRowWithSchema}
import org.apache.spark.sql.types.{DataType, StructType}

case class CustomData(value: String)

class CustomEncoder extends ExpressionEncoder[CustomData] {
  override def schema: StructType = {
    new StructType().add("value", StringType)
  }

  override def bind(child: Expression): Encoder[CustomData] = {
    this
  }

  override def createDeserializer(): Expression = {
    val dataType = schema.toAttributes.head.dataType
    val converter = CatalystTypeConverters.createToScalaConverter(dataType)
    val row = new GenericRowWithSchema(Array.empty, schema)
    val deserializer = CatalystTypeConverters.createDeserializer(dataType, row.schema)
    deserializer(converter(row))
  }

  override def createSerializer(): Expression = {
    val dataType = schema.toAttributes.head.dataType
    val converter = CatalystTypeConverters.createToCatalystConverter(dataType)
    val serializer = CatalystTypeConverters.createSerializer(dataType)
    serializer(converter(new CustomData("")))
  }
}

val customEncoder = new CustomEncoder()
val customDataframe = spark.createDataFrame(Seq(CustomData("example")), customEncoder.schema)
val encodedDataframe = customEncoder.toRow(customDataframe)