如何创建AVRODeserializationSchema并在flink kafka消费者中使用?

tvokkenx  于 2021-06-21  发布在  Flink
关注(0)|答案(1)|浏览(573)

我试图创建一个Kafkaavro系列主题的flink消费者。我有Kafka主题流avro序列化数据。我可以通过avroconsoleconsumer看到。
flink1.6.0添加了一个avrodeserializationschema,但我找不到完整的用法示例。是的,有一些生成了avrodeserialization类,似乎是在1.6.0之前添加的类。
我有一个通过avro工具生成的avro类。
现在,我一直在努力遵循现有的例子,但他们是不同的,我不能让事情继续下去(我不经常用java编程)
大多数使用以下形式

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

其中myclass是通过avro工具jar生成的avro类。走这条路对吗?在执行此操作和利用内部flink 1.6.0 avrodeserializationschema类时,我遇到了一些私有/公共访问问题。我必须创建一个新类并扩展avrodeserializationschema吗?

66bbxpm5

66bbxpm51#

好的,我深入研究了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");
}

}

相关问题