Flink使用 KafkaSource消费 Kafka中的数据

这篇具有很好参考价值的文章主要介绍了Flink使用 KafkaSource消费 Kafka中的数据。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

1.前言

目前，很多 flink相关的书籍和网上的文章讲解如何对接 kafka时都是使用的 FlinkKafkaConsumer，如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Properties properties = new Properties();
//指定kafka的Broker地址
properties.setProperty("bootstrap.servers", "192.168.xx.xx:9092");
//指定组ID
properties.setProperty("group.id", "flink-demo");
//如果没有记录偏移量，第一次从最开始消费
properties.setProperty("auto.offset.reset", "earliest");

FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>("topic005", new SimpleStringSchema(), properties);

新版的 flink，比如 1.14.3已经将 FlinkKafkaConsumer标记为 deprecated（不推荐）,如下：

'org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer' is deprecated

2.如何使用 KafkaSource

新版本的 flink应该使用 KafkaSource来消费 kafka中的数据，详细代码如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
KafkaSource<String> kfkSource = KafkaSource.<String>builder()
    .setBootstrapServers("192.168.xx.xx:9092")
    .setGroupId("flink-demo")
    .setTopics("topic005")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();
DataStreamSource<String> lines = env.fromSource(kfkSource, WatermarkStrategy.noWatermarks(), "kafka source");