
<!-- 引入 kafka连接器依赖-->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>1.17.0</version>
</dependency>

2、API使用说明

官网链接：Apache Kafka 连接器

语法说明:


// 1.初始化 KafkaSource 实例
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)                           // 必填：指定broker连接信息 (为保证高可用,建议多指定几个节点)                     
    .setTopics("input-topic")                               // 必填：指定要消费的topic
    .setGroupId("my-group")                                 // 必填：指定消费者的groupid(不存在时会自动创建)
    .setValueOnlyDeserializer(new SimpleStringSchema())     // 必填：指定反序列化器(用来解析kafka消息数据，转换为flink数据类型)
    .setStartingOffsets(OffsetsInitializer.earliest())      // 可选：指定启动任务时的消费位点（不指定时，将默认使用 OffsetsInitializer.earliest()）
    .build(); 
 
// 2.通过 fromSource + KafkaSource 获取 DataStreamSource
env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

3、这是一个完整的入门案例

开发语言：java1.8

flink版本：flink1.17.0


public class ReadKafka {
    public static void main(String[] args) throws Exception {
        newAPI();
    }
 
    public static void newAPI() throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        // 2.读取kafka数据
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("worker01:9092")               // 必填：指定broker连接信息 (为保证高可用,建议多指定几个节点)
                .setTopics("20230810")                              // 必填：指定要消费的topic
                .setGroupId("FlinkConsumer")                        // 必填：指定消费者的groupid(不存在时会自动创建)
                .setValueOnlyDeserializer(new SimpleStringSchema()) // 必填：指定反序列化器(用来解析kafka消息数据)
                .setStartingOffsets(OffsetsInitializer.earliest())  // 可选：指定启动任务时的消费位点（不指定时，将默认使用 OffsetsInitializer.earliest()）
                .build();
 
        env.fromSource(source,
                WatermarkStrategy.noWatermarks(),
                "Kafka Source")
                .print()
        ;
 
        // 3.触发程序执行
        env.execute();
    }
}

4、Kafka消息应该如何解析

代码中需要提供一个反序列化器（Deserializer）来对 Kafka 的消息进行解析

反序列化器的功能：

将Kafka ConsumerRecords转换为Flink处理的数据类型(Java/Scala对象)

反序列化器通过 setDeserializer(KafkaRecordDeserializationSchema.of(反序列化器类型)) 指定

下面介绍两种常用Kafka消息解析器：

KafkaRecordDeserializationSchema.of(new JSONKeyValueDeserializationSchema(true)) ：

1、返回完整的Kafka消息，将JSON字符串反序列化为ObjectNode对象

2、可以选择是否返回Kafak消息的Metadata信息，true-返回，false-不返回

KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class) ：

1、只返回Kafka消息中的value部分

4.1、只获取Kafka消息的value部分

4.2、获取完整Kafka消息(key、value、Metadata)

kafak消息格式：

key = {"nation":"蜀国"}

value = {"ID":整数}


    public static void ParseMessageJSONKeyValue() throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        // 2.读取kafka数据
        KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
                .setBootstrapServers("worker01:9092")               // 必填：指定broker连接信息 (为保证高可用,建议多指定几个节点)
                .setTopics("9527")                                  // 必填：指定要消费的topic
                .setGroupId("FlinkConsumer")                        // 必填：指定消费者的groupid(不存在时会自动创建)
                // 必填：指定反序列化器(将kafak消息解析为ObjectNode，json对象)
                .setDeserializer(KafkaRecordDeserializationSchema.of(
                        // includeMetadata = (true:返回Kafak元数据信息 false:不返回)
                        new JSONKeyValueDeserializationSchema(true)
                ))
                .setStartingOffsets(OffsetsInitializer.latest())  // 可选：指定启动任务时的消费位点（不指定时，将默认使用 OffsetsInitializer.earliest()）
                .build();
 
        env
                .fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")
                .print()
        ;
 
        // 3.触发程序执行
        env.execute();
 
    }

运行结果：

常见报错：


Caused by: java.io.IOException: Failed to deserialize consumer record ConsumerRecord(topic = 9527, partition = 0, leaderEpoch = 0, offset = 1064, CreateTime = 1691668775938, serialized key size = 4, serialized value size = 9, headers = RecordHeaders(headers = [], isReadOnly = false), key = [B@5e9eaab8, value = [B@67390400).
	at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:57)
	at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53)
	... 14 more
Caused by: org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'xxxx': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (byte[])"xxxx"; line: 1, column: 5]

报错原因：

出现这个报错，一般是使用flink读取fafka时，使用JSONKeyValueDeserializationSchema

来解析消息时，kafka消息中的key 或者 value 内容不符合json格式而造成的解析错误

例如下面这个格式，就会造成解析错误 key=1000，value=你好

那应该怎么解决呢？

1、如果有权限修改Kafka消息格式，可以将Kafka消息key&value内容修改为Json格式

2、如果没有权限修改Kafka消息格式(比如线上环境，修改比较困难)，可以重新实现

       JSONKeyValueDeserializationSchema类，根据所需格式来解析Kafka消息(可以参考源码)

4.3、自定义Kafka消息解析器

生产中对Kafka消息及解析的格式总是各种各样的，当flink预定义的解析器满足不了业务需求时，可以通过自定义kafka消息解析器来完成业务的支持

例如，当使用 MyJSONKeyValueDeserializationSchema 获取Kafka元数据时，只返回了 offset、topic、partition 三个字段信息，现在需要`kafka生产者写入数据时的timestamp`，就可以通过自定义kafka消息解析器来完成

代码示例：


// TODO 自定义Kafka消息解析器，在 metadata 中增加 timestamp字段
public class MyJSONKeyValueDeserializationSchema implements KafkaDeserializationSchema<ObjectNode>{
 
        private static final long serialVersionUID = 1509391548173891955L;
 
        private final boolean includeMetadata;
        private ObjectMapper mapper;
 
        public MyJSONKeyValueDeserializationSchema(boolean includeMetadata) {
            this.includeMetadata = includeMetadata;
        }
 
        @Override
        public void open(DeserializationSchema.InitializationContext context) throws Exception {
            mapper = JacksonMapperFactory.createObjectMapper();
        }
 
        @Override
        public ObjectNode deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
            ObjectNode node = mapper.createObjectNode();
            if (record.key() != null) {
                node.set("key", mapper.readValue(record.key(), JsonNode.class));
            }
            if (record.value() != null) {
                node.set("value", mapper.readValue(record.value(), JsonNode.class));
            }
            if (includeMetadata) {
                node.putObject("metadata")
                        .put("offset", record.offset())
                        .put("topic", record.topic())
                        .put("partition", record.partition())
                        // 添加 timestamp 字段
                        .put("timestamp",record.timestamp())
                ;
            }
            return node;
        }
 
        @Override
        public boolean isEndOfStream(ObjectNode nextElement) {
            return false;
        }
 
        @Override
        public TypeInformation<ObjectNode> getProducedType() {
            return getForClass(ObjectNode.class);
        }
    }

运行结果：

5、起始消费位点应该如何设置

起始消费位点说明：

起始消费位点是指 启动flink任务时，应该从哪个位置开始读取Kafka的消息

下面介绍下常用的三个设置：

OffsetsInitializer.earliest() ：

从最早位点开始消

这里的最早指的是Kafka消息保存的时长(默认为7天，生成环境各公司略有不同)

该这设置为默认设置，当不指定OffsetsInitializer.xxx时，默认为earliest()

OffsetsInitializer.latest() ：

从最末尾位点开始消费

这里的最末尾指的是flink任务启动时间点之后生产的消息

OffsetsInitializer.timestamp(时间戳) ：

从时间戳大于等于指定时间戳（毫秒）的数据开始消费

下面用案例说明下，三种设置的效果，kafak生成10条数据，如下：

5.1、earliest()

代码示例：


KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
        .setBootstrapServers("worker01:9092")
        .setTopics("23230811")
        .setGroupId("FlinkConsumer")
        // 将kafka消息解析为Json对象，并返回元数据
        .setDeserializer(KafkaRecordDeserializationSchema.of(
                new JSONKeyValueDeserializationSchema(true)
        ))
        // 设置起始消费位点：从最早位置开始消费（该设置为默认设置）
        .setStartingOffsets(OffsetsInitializer.earliest())
        .build();

运行结果：

5.2、latest()

代码示例：


KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
        .setBootstrapServers("worker01:9092")
        .setTopics("23230811")
        .setGroupId("FlinkConsumer")
        // 将kafka消息解析为Json对象，并返回元数据
        .setDeserializer(KafkaRecordDeserializationSchema.of(
                new JSONKeyValueDeserializationSchema(true)
        ))
        // 设置起始消费位点：从最末尾位点开始消费
        .setStartingOffsets(OffsetsInitializer.latest())
        .build();

运行结果：

5.3、timestamp()

代码示例：


KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
        .setBootstrapServers("worker01:9092")
        .setTopics("23230811")
        .setGroupId("FlinkConsumer")
        // 将kafka消息解析为Json对象，并返回元数据
        .setDeserializer(KafkaRecordDeserializationSchema.of(
                new MyJSONKeyValueDeserializationSchema(true)
        ))
        // 设置起始消费位点：从指定时间戳后开始消费
        .setStartingOffsets(OffsetsInitializer.timestamp(1691722791273L))
        .build();

运行结果：

6、Kafka分区扩容了，该怎么办 —— 动态分区检查

在flink1.13的时候，如果Kafka分区扩容了，只有通过重启flink任务，才能消费到新增分区的数据，小编就曾遇到过上游业务部门的kafka分区扩容了，并没有通知下游使用方，导致实时指标异常，甚至丢失了数据。

在flink1.17的时候，可以通过`开启动态分区检查`，来实现不用重启flink任务，就能消费到新增分区的数据

开启分区检查：(默认不开启)


KafkaSource.builder()
    .setProperty("partition.discovery.interval.ms", "10000"); // 每 10 秒检查一次新分区

代码示例：


KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
        .setBootstrapServers("worker01:9092")
        .setTopics("9527")
        .setGroupId("FlinkConsumer")
        // 将kafka消息解析为Json对象，并返回元数据
        .setDeserializer(KafkaRecordDeserializationSchema.of(
                new JSONKeyValueDeserializationSchema(true)
        ))
        // 设置起始消费位点：从最末尾位点开始消费
        .setStartingOffsets(OffsetsInitializer.latest())
        // 开启动态分区检查（默认不开启）
        .setProperty("partition.discovery.interval.ms", "10000") // 每 10 秒检查一次新分区
        .build();

7、在加载KafkaSource时提取事件时间&添加水位线

可以在 fromSource(source,WatermarkStrategy,sourceName) 时，提取事件时间和制定水位线生成策略

注意：当不指定事件时间提取器时，Kafka Source 使用 Kafka 消息中的时间戳作为事件时间

7.1、使用内置的单调递增的水位线生成器 + kafka timestamp 为事件时间

代码示例：


    // 在读取Kafka消息时，提取事件时间&插入水位线
    public static void KafkaSourceExtractEventtimeAndWatermark() throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        // 2.读取kafka数据
        KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
                .setBootstrapServers("worker01:9092")
                .setTopics("9527")
                .setGroupId("FlinkConsumer")
                // 将kafka消息解析为Json对象，并返回元数据
                .setDeserializer(KafkaRecordDeserializationSchema.of(
                        new MyJSONKeyValueDeserializationSchema(true)
                ))
                // 设置起始消费位点：从最末尾位点开始消费
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();
 
        env.fromSource(source,
                        // 使用内置的单调递增的水位线生成器（默认使用 kafka的timestamp作为事件时间）
                        WatermarkStrategy.forMonotonousTimestamps(),
                        "Kafka Source")
                // 通过 ProcessFunction 查看提取的事件时间和水位线信息
                .process(
                        new ProcessFunction<ObjectNode, String>() {
                            @Override
                            public void processElement(ObjectNode kafkaJson, ProcessFunction<ObjectNode, String>.Context ctx, Collector<String> out) throws Exception {
                                // 当前处理时间
                                long currentProcessingTime = ctx.timerService().currentProcessingTime();
                                // 当前水位线
                                long currentWatermark = ctx.timerService().currentWatermark();
                                StringBuffer record = new StringBuffer();
                                record.append("========================================\n");
                                record.append(kafkaJson + "\n");
                                record.append("currentProcessingTime：" + currentProcessingTime + "\n");
                                record.append("currentWatermark：" + currentWatermark + "\n");
                                record.append("kafka-ID：" + Long.parseLong(kafkaJson.get("value").get("ID").toString()) + "\n");
                                record.append("kafka-timestamp：" + Long.parseLong(kafkaJson.get("metadata").get("timestamp").toString()) + "\n");
                                out.collect(record.toString());
 
                            }
                        }
                ).print();
 
        // 3.触发程序执行
        env.execute();
    }

运行结果：

7.2、使用内置的单调递增的水位线生成器 + kafka 消息中的 ID字段为事件时间

代码示例：


    // 在读取Kafka消息时，提取事件时间&插入水位线
    public static void KafkaSourceExtractEventtimeAndWatermark() throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 
        // 2.读取kafka数据
        KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
                .setBootstrapServers("worker01:9092")
                .setTopics("9527")
                .setGroupId("FlinkConsumer")
                // 将kafka消息解析为Json对象，并返回元数据
                .setDeserializer(KafkaRecordDeserializationSchema.of(
                        new MyJSONKeyValueDeserializationSchema(true)
                ))
                // 设置起始消费位点：从最末尾位点开始消费
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();
 
        env.fromSource(source,
                        // 使用内置的单调递增的水位线生成器（使用 kafka消息中的ID字段作为事件时间）
                        WatermarkStrategy.<ObjectNode>forMonotonousTimestamps()
                                // 提取 Kafka消息中的 ID字段作为 事件时间
                                .withTimestampAssigner(
                                        (json, timestamp) -> Long.parseLong(json.get("value").get("ID").toString())
                                ),
 
                        "Kafka Source")
                // 通过 ProcessFunction 查看提取的事件时间和水位线信息
                .process(
                        new ProcessFunction<ObjectNode, String>() {
                            @Override
                            public void processElement(ObjectNode kafkaJson, ProcessFunction<ObjectNode, String>.Context ctx, Collector<String> out) throws Exception {
                                // 当前处理时间
                                long currentProcessingTime = ctx.timerService().currentProcessingTime();
                                // 当前水位线
                                long currentWatermark = ctx.timerService().currentWatermark();
                                StringBuffer record = new StringBuffer();
                                record.append("========================================\n");
                                record.append(kafkaJson + "\n");
                                record.append("currentProcessingTime：" + currentProcessingTime + "\n");
                                record.append("currentWatermark：" + currentWatermark + "\n");
                                record.append("kafka-ID：" + Long.parseLong(kafkaJson.get("value").get("ID").toString()) + "\n");
                                record.append("kafka-timestamp：" + Long.parseLong(kafkaJson.get("metadata").get("timestamp").toString()) + "\n");
                                out.collect(record.toString());
 
                            }
                        }
                ).print();
 
        // 3.触发程序执行
        env.execute();
    }

运行结果：

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop】

4.3、Flink任务怎样读取Kafka中的数据_flink读取kafka数据

1、添加pom依赖

2、API使用说明

3、这是一个完整的入门案例

4、Kafka消息应该如何解析

4.1、只获取Kafka消息的value部分

4.2、获取完整Kafka消息(key、value、Metadata)

4.3、自定义Kafka消息解析器

5、起始消费位点应该如何设置

5.1、earliest()

5.2、latest()

5.3、timestamp()

6、Kafka分区扩容了，该怎么办 —— 动态分区检查

7、在加载KafkaSource时提取事件时间&添加水位线

7.1、使用内置的单调递增的水位线生成器 + kafka timestamp 为事件时间

7.2、使用内置的单调递增的水位线生成器 + kafka 消息中的 ID字段 为事件时间

7.2、使用内置的单调递增的水位线生成器 + kafka 消息中的 ID字段为事件时间