Flink ParquetAvroWriters 给出演员表错误答案

【问题标题】：Flink ParquetAvroWriters giving cast ERRORFlink ParquetAvroWriters 给出演员表错误
【发布时间】：2020-01-19 06:03:43
【问题描述】：

在从 Kafka 读取 GenericRecord 后，我编写了一个示例代码以将流写入 parquet 格式

 Properties config = new Properties();
        config.setProperty("bootstrap.servers", "localhost:9092");
        config.setProperty("group.id", "1");
        config.setProperty("zookeeper.connect", "localhost:2181");
        String schemaRegistryUrl = "http://127.0.0.1:8081";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        File file = new File(EventProcessor.class.getClassLoader().getResource("event.avsc").getFile());
        Schema schema = new Schema.Parser().parse(file);

        DataStreamSource<GenericRecord> input = env
                .addSource(
                        new FlinkKafkaConsumer010<GenericRecord>("event_new",
                                new KafkaGenericAvroDeserializationSchema(schemaRegistryUrl),
                                config).setStartFromEarliest());


        Path path = new Path("/tmp");

        final StreamingFileSink sink = StreamingFileSink.forBulkFormat
                (path, ParquetAvroWriters.forGenericRecord(schema)).build();

        input.addSink(sink);

当我运行此代码时出现错误：

Caused by: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator

Caused by: java.lang.ClassCastException: org.apache.avro.util.Utf8 cannot be cast to org.apache.avro.generic.IndexedRecord
    at org.apache.avro.generic.GenericData.getField(GenericData.java:697)
    at org.apache.parquet.avro.AvroWriteSupport.writeRecordFields(AvroWriteSupport.java:188)

我无法理解出了什么问题。请帮助我理解并解决这个问题。

【问题讨论】：

标签： apache-flink avro parquet

【解决方案1】：

最可能的原因是您的 event.avsc 与存储在 Kafka 中的记录不匹配。它正在寻找一个字符串，它需要一条记录。

如果您从 Kafka 添加架构和示例记录（例如使用控制台消费者打印），那么我可以提供更多帮助。

【讨论】：

我已经解决了。只是因为架构错误。