如何创建 AvroDeserialzationSchema 并在 Flink Kafka Consumer 中使用？答案

【问题标题】：How do I create the AvroDeserialzationSchema and use in a Flink Kafka Consumer?如何创建 AvroDeserialzationSchema 并在 Flink Kafka Consumer 中使用？
【发布时间】：2018-08-20 02:30:50
【问题描述】：

我正在尝试为 kafka avro 序列化主题创建一个 flink 消费者。我有 kafka 主题流式传输 avro 序列化数据。我可以通过 avroconsoleconsumer 看到它。

Flink 1.6.0 添加了AvroDeserializationSchema，但我找不到完整的用法示例。是的，有一些似乎在 1.6.0 之前生成了一个 avrodeserialization 类。

我有一个通过 avro-tools 生成的 avro 类。

现在我一直在尝试遵循现有的示例，但它们的不同之处足以让我无法继续进行。（我不经常用 Java 编程）

大多数使用以下某种形式

Myclass mc = new MyClass();
AvroDeserializationSchema<Myclass> ads = new AvroDeserializationSchema<> (Myclass.class);
FlinkKafkaConsumer010<Myclass> kc = new FlinkKafkaConsumer010<>(topic,ads,properties);

其中 Myclass 是通过 avro-tools jar 生成的 avro 类。这是正确的方法吗？在执行此操作并利用内部 flink 1.6.0 avrodeserializationschema 类时，我遇到了一些私有/公共访问问题。我是否必须创建一个新类并扩展 avrodeserializationschema？

【问题讨论】：

我认为这个问题与flink无关。您应该在 kafkajdbcconnector 项目文档中寻找答案。
我编辑了这个问题，因为我通过 flink avrodeserializationschema 和 flinkkafkaconsumer 类将其缩小到 avrodeserialization。

标签： java apache-flink avro-tools

【解决方案1】：

好的，我研究了 kafka 消费者 javadocs 并找到了一个提取消费 avro 流的示例。我仍然需要将 kafka 消费转换为 flinkKafkaConsumer，但下面的代码可以工作。

对于工作的 io.confluent 引用，我必须添加存储库和对 pom 文件的依赖项。

<repository>
  <id>confluent</id>
    <url>http://packages.confluent.io/maven/</url>
</repository>

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>3.1.1</version>
</dependency>

public class StreamingJob {

//  static  DeserializationSchema<pendingsv> avroSchema = new AvroDeserializationSchema<pendingsv>(pendingsv.class);
    public static void main(String[] args) throws Exception {
        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "opssupport.alarms");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroDeserializer");
        props.put("schema.registry.url", "http://localhost:8081");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        String topic = "pendingSVs_";
        final Consumer<String, GenericRecord> consumer = new KafkaConsumer<String, GenericRecord>(props);
        consumer.subscribe(Arrays.asList(topic));

        try {
            while (true) {
                ConsumerRecords<String, GenericRecord> records = consumer.poll(100);
                for (ConsumerRecord<String, GenericRecord> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s \n", record.offset(), record.key(), record.value());
                }
            }
        } finally {
            consumer.close();
        }

        // execute program
        //env.execute("Flink Streaming Java API Skeleton");
    }
}

【讨论】：