如何在Spark中将结构编码成Avro记录？

libraryDependencies += "org.apache.avro" % "avro" % "1.10.2"

import org.apache.spark.sql.{SparkSession, SaveMode}
import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericRecord}
import org.apache.avro.io.EncoderFactory
import org.apache.avro.specific.SpecificDatumWriter

val spark = SparkSession.builder()
  .appName("AvroEncoder")
  .master("local")
  .getOrCreate()

val schemaString =
  """
    |{
    |  "type": "record",
    |  "name": "Person",
    |  "fields": [
    |    {"name": "name", "type": "string"},
    |    {"name": "age", "type": "int"},
    |    {"name": "email", "type": "string"}
    |  ]
    |}
  """.stripMargin

val schema = new Schema.Parser().parse(schemaString)

val data = Seq(("John", 30, "john@example.com"), ("Alice", 25, "alice@example.com"))
val df = spark.createDataFrame(data).toDF("name", "age", "email")

val avroRecords = df.rdd.map { row =>
  val record = new GenericData.Record(schema)
  record.put("name", row.getAs[String]("name"))
  record.put("age", row.getAs[Int]("age"))
  record.put("email", row.getAs[String]("email"))
  record
}

val encoderFactory = EncoderFactory.get()
val avroEncoder = encoderFactory.binaryEncoder(System.out, null)
val writer = new SpecificDatumWriter[GenericRecord](schema)

avroRecords.foreach { record =>
  writer.write(record, avroEncoder)
  avroEncoder.flush()
}