如何从Kafka Topic导出Avro生产数据并在Java/Scala中读取？

It sounds like you're hitting common pitfalls with Confluent Avro-formatted Kafka data—let's break down reliable methods to export the data properly and then read it in your Java/Scala applications.

Fixing the kafka-avro-console-consumer Output #

Your issue with incomplete JSON is almost certainly due to default truncation of long values or missing configuration to handle both keys and values (if your topic uses Avro for keys too). Here's a full command to export complete, valid JSON:

kafka-avro-console-consumer \
  --bootstrap-server <KAFKA_BROKER>:9092 \
  --topic <YOUR_TOPIC> \
  --from-beginning \
  --property schema.registry.url=http://<SCHEMA_REGISTRY>:8081 \
  --property print.key=true \
  --property key.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer \
  --property value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer \
  --property truncate=false \
  --property format=json \
  > exported_full_data.json

• truncate=false ensures long strings/fields aren't cut off
• Explicitly setting key/value deserializers guarantees both parts of the message are processed (critical if your topic uses Avro for keys)
• Redirecting to a file gives you clean, line-delimited JSON perfect for testing

More Robust Export with Kafka Connect #

For larger datasets or automated exports, Kafka Connect's FileStreamSinkConnector is a better choice—it handles bulk data reliably and avoids console consumer limitations.

1. Create a connector config file (file-sink-config.properties):

name=avro-file-sink
connector.class=org.apache.kafka.connect.file.FileStreamSinkConnector
tasks.max=1
topics=<YOUR_TOPIC>
file=exported_avro_data.json
# Configure Avro converters for keys and values
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://<SCHEMA_REGISTRY>:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://<SCHEMA_REGISTRY>:8081
# Disable schema embedding in each line (optional, keeps output cleaner)
value.converter.schemas.enable=false

2. Run the connector using Connect standalone mode:

connect-standalone <PATH_TO_CONNECT_STANDALONE_CONFIG> file-sink-config.properties

Reading the Exported JSON in Java/Scala #

Once you have valid JSON, you can read it using Avro's built-in tools or generated classes for type safety.

Java Example with Generated Avro Classes #

First, generate classes from your topic's schema (fetch it from the registry):

# Get the latest schema for your topic's value
curl http://<SCHEMA_REGISTRY>:8081/subjects/<YOUR_TOPIC>-value/versions/latest > schema.json
# Compile to Java classes
avro-tools compile schema schema.json com.your.project.model

Then read the exported JSON:

import com.your.project.model.YourRecord;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import java.io.FileReader;
import java.io.IOException;

public class AvroJsonReader {
    public static void main(String[] args) throws IOException {
        SpecificDatumReader<YourRecord> reader = new SpecificDatumReader<>(YourRecord.class);
        Decoder decoder = DecoderFactory.get().jsonDecoder(YourRecord.getClassSchema(), new FileReader("exported_full_data.json"));
        
        YourRecord record;
        while ((record = reader.read(null, decoder)) != null) {
            // Access fields directly with type safety
            System.out.println("Field value: " + record.getYourField());
        }
    }
}

Scala Example (Using Avro4s for Simplicity) #

If you prefer Scala, the avro4s library simplifies working with Avro data:

import com.sksamuel.avro4s.AvroInputStream
import java.io.File

case class YourRecord(field1: String, field2: Int) // Match your schema structure

object AvroJsonReader {
  def main(args: Array[String]): Unit = {
    val input = AvroInputStream.json[YourRecord].from(new File("exported_full_data.json")).build()
    input.foreach(record => println(s"Processed: $record"))
    input.close()
  }
}

Why Your Original Approaches Failed #

• String/Binary Deserializer: Confluent's Avro serializer prepends a magic byte and schema ID to each message. Using these deserializers reads that raw metadata as part of the payload, making it unparseable by standard Avro tools.
• Incomplete JSON from kafka-avro-console-consumer: Default settings truncate long fields to keep console output readable—adding truncate=false fixes this.

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