将Spark DataFrame行转换为Avro并发布至Kafka及流式传输实现咨询

Hey there! Let's break down exactly how to solve your problem—converting a Spark DataFrame (including streaming data) to Avro format and sending it to Kafka, with full support for your specific schema.

1. 先搞定依赖 #

First, make sure your Spark setup has the required connectors. If you're using spark-submit, include these packages (adjust versions to match your Spark release):

spark-submit \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0,org.apache.spark:spark-avro_2.12:3.3.0 \
  your-app.jar

If you're using a build tool like Maven, add these dependencies to your pom.xml:

<dependencies>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
    <version>3.3.0</version>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-avro_2.12</artifactId>
    <version>3.3.0</version>
    <scope>provided</scope>
  </dependency>
</dependencies>

2. 定义匹配的Avro Schema #

Your DataFrame has nested nullable fields, so we need an Avro Schema that mirrors this structure (including null unions for nullable columns). Here's the exact schema you'll need:

{
  "type": "record",
  "name": "CustomerCheckRecord",
  "fields": [
    {"name": "performCheck", "type": ["null", "string"]},
    {"name": "clientTag", "type": ["null", {
      "type": "record",
      "name": "ClientTag",
      "fields": [{"name": "key", "type": ["null", "string"]}]
    }]},
    {"name": "contactPoint", "type": ["null", {
      "type": "record",
      "name": "ContactPoint",
      "fields": [
        {"name": "email", "type": ["null", "string"]},
        {"name": "type", "type": ["null", "string"]}
      ]
    }]}
  ]
}

Store this schema as a string in your code (let's call it avroSchemaString) for later use.

3. 流式处理：DataFrame转Avro并写入Kafka #

Assuming you already have a streaming DataFrame (e.g., read from Kafka, files, or another source), here's how to convert it to Avro and send it to Kafka:

Step 3.1: Import required functions #

import org.apache.spark.sql.functions._
import org.apache.spark.sql.avro.functions.to_avro

Step 3.2: Transform the streaming DataFrame #

We need to format the data to fit Kafka's expected structure (a key and value column). The value will be our Avro-serialized binary data:

// Replace `streamingDF` with your actual streaming DataFrame
val avroKafkaDF = streamingDF
  .select(
    // Optional: Use a field as the Kafka key (e.g., clientTag.key)
    col("clientTag.key").cast("string").alias("key"),
    // Serialize the entire row to Avro using our predefined schema
    to_avro(
      struct("performCheck", "clientTag", "contactPoint"),
      avroSchemaString
    ).alias("value")
  )

Step 3.3: Write to Kafka stream #

val streamingQuery = avroKafkaDF
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-broker-1:9092,your-broker-2:9092")
  .option("topic", "your-target-topic")
  // Use "append" mode for most streaming use cases (adds new records)
  .outputMode("append")
  // Critical: Specify a checkpoint directory for fault tolerance
  .option("checkpointLocation", "/path/to/your/checkpoint/dir")
  .start()

// Keep the stream running
streamingQuery.awaitTermination()

4. 批处理/单个Row转Avro（如果需要） #

If you need to convert individual Spark Row objects to Avro (e.g., in a UDF or batch job), use the Avro Java API directly:

Step 4.1: Helper functions to convert Row to Avro #

import org.apache.avro.Schema
import org.apache.avro.generic.{GenericData, GenericRecord}
import org.apache.avro.io.{DatumWriter, EncoderFactory}
import org.apache.avro.generic.GenericDatumWriter
import java.io.ByteArrayOutputStream

// Parse the Avro Schema
val avroSchema = new Schema.Parser().parse(avroSchemaString)

// Convert Spark Row to Avro GenericRecord
def rowToAvroRecord(row: Row): GenericRecord = {
  val record = new GenericData.Record(avroSchema)
  
  // Handle top-level fields
  record.put("performCheck", row.getAs[String]("performCheck"))
  
  // Handle nested clientTag struct
  val clientTagRow = row.getAs[Row]("clientTag")
  if (clientTagRow != null) {
    val clientTagRecord = new GenericData.Record(avroSchema.getField("clientTag").schema().getTypes.get(1))
    clientTagRecord.put("key", clientTagRow.getAs[String]("key"))
    record.put("clientTag", clientTagRecord)
  }
  
  // Handle nested contactPoint struct
  val contactPointRow = row.getAs[Row]("contactPoint")
  if (contactPointRow != null) {
    val contactPointRecord = new GenericData.Record(avroSchema.getField("contactPoint").schema().getTypes.get(1))
    contactPointRecord.put("email", contactPointRow.getAs[String]("email"))
    contactPointRecord.put("type", contactPointRow.getAs[String]("type"))
    record.put("contactPoint", contactPointRecord)
  }
  
  record
}

// Serialize GenericRecord to binary Avro data
def serializeAvroRecord(record: GenericRecord): Array[Byte] = {
  val out = new ByteArrayOutputStream()
  val writer: DatumWriter[GenericRecord] = new GenericDatumWriter[GenericRecord](avroSchema)
  val encoder = EncoderFactory.get().binaryEncoder(out, null)
  writer.write(record, encoder)
  encoder.flush()
  out.toByteArray()
}

Step 4.2: Usage example #

// Sample Row matching your schema
val sampleRow = Row(
  Row("value"),
  Row("abc@gmail.com", "EML"),
  "N"
)

// Convert to Avro binary
val avroBytes = serializeAvroRecord(rowToAvroRecord(sampleRow))

You can then send this binary data to Kafka using a Kafka producer client.

Key Notes to Avoid Issues #

• Version Compatibility: Ensure Spark Avro and Kafka connector versions match your Spark cluster version.
• Null Handling: Always include ["null", ...] in your Avro Schema for nullable DataFrame columns—this prevents serialization failures.
• Checkpointing: Never skip the checkpointLocation for streaming jobs; it ensures your stream can recover from failures.
• Kafka Permissions: Verify your Spark cluster has network access to Kafka brokers and write permissions for the target topic.

内容的提问来源于stack exchange，提问作者Vinil