Flink ParquetAvroWriters抛出类型转换错误，请求排查解决

这个错误我之前也碰到过，本质是反序列化后得到的对象不是预期的GenericRecord，而是Utf8类型，导致Parquet写入器无法正确处理。咱们一步步来排查解决：

错误原因分析 #

从堆栈信息来看，AvroWriteSupport在尝试写入记录字段时，发现拿到的是Utf8对象而非IndexedRecord（GenericRecord是它的子类）。这说明：

• 要么Kafka中的消息本身就不是符合你定义的Avro Schema的完整记录，可能只是单纯的字符串或者单个字段；
• 要么反序列化器没有正确把Kafka消息解析成GenericRecord。

排查与解决方案 #

1. 先验证Kafka消息的实际格式 #

首先得确认Kafka里的消息到底是什么样的，用Kafka自带的工具来查看：

• 如果是查看字符串格式的消息：

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic event_new --from-beginning --property value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

• 如果是查看Avro格式的消息（需要schema registry）：

kafka-avro-console-consumer.sh --bootstrap-server localhost:9092 --topic event_new --from-beginning --schema-registry-url http://127.0.0.1:8081

看看消息结构是否和你本地event.avsc定义的一致。如果消息只是单纯的字符串，那反序列化器自然会解析成Utf8类型。

2. 调整反序列化器配置 #

如果Kafka消息确实是Avro格式，那问题可能出在KafkaGenericAvroDeserializationSchema的使用上：

• 你当前的代码只传入了schema registry的URL，没有指定本地schema，可能registry里的schema和你本地的event.avsc不匹配。可以尝试直接把本地解析的schema传给反序列化器：

DataStreamSource<GenericRecord> input = env
    .addSource(
        new FlinkKafkaConsumer010<GenericRecord>(
            "event_new", 
            new KafkaGenericAvroDeserializationSchema(schema), // 传入本地解析好的schema
            config).setStartFromEarliest());

• 另外，确认schema registry中该topic对应的schemaID是否正确，消息中的schemaID是否和registry里的匹配。

3. 处理字符串格式的消息（如果适用） #

如果Kafka里的消息是JSON格式的Avro字符串（而非二进制Avro），那需要先把字符串转成GenericRecord再写入Parquet：

// 先读取字符串
DataStreamSource<String> input = env
    .addSource(
        new FlinkKafkaConsumer010<String>(
            "event_new", 
            new SimpleStringSchema(), 
            config).setStartFromEarliest());

// 转换为GenericRecord
DataStream<GenericRecord> avroStream = input.map(new MapFunction<String, GenericRecord>() {
    @Override
    public GenericRecord map(String value) throws Exception {
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
        try (StringReader stringReader = new StringReader(value)) {
            Decoder decoder = DecoderFactory.get().jsonDecoder(schema, stringReader);
            return reader.read(null, decoder);
        }
    }
});

// 写入Parquet
avroStream.addSink(sink);

4. 确认Parquet写入器的Schema匹配 #

最后检查ParquetAvroWriters.forGenericRecord(schema)传入的schema，是否和反序列化得到的GenericRecord的schema完全一致，字段名、类型都不能有差异。

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