4.3、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字段 为事件时间

---

1、添加pom依靠

我们可以利用Flink官方提供毗连Kafka的工具flink-connector-kafka
该工具实现了一个消费者FlinkKafkaConsumer，可以用它来读取kafka的数据
如果想利用这个通用的Kafka毗连工具，需要引入jar依靠

1. <!-- 引入 kafka连接器依赖-->
2. <dependency>
3.     <groupId>org.apache.flink</groupId>
4.     <artifactId>flink-connector-kafka</artifactId>
5.     <version>1.17.0</version>
6. </dependency>

复制代码

---

2、API利用说明

官网链接：Apache Kafka 毗连器
语法说明: 

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

复制代码

---

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

开发语言：java1.8
flink版本：flink1.17.0

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

复制代码

---

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":整数}

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

复制代码

运行效果：    

常见报错： 

1. 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).
2.         at org.apache.flink.connector.kafka.source.reader.deserializer.KafkaDeserializationSchemaWrapper.deserialize(KafkaDeserializationSchemaWrapper.java:57)
3.         at org.apache.flink.connector.kafka.source.reader.KafkaRecordEmitter.emitRecord(KafkaRecordEmitter.java:53)
4.         ... 14 more
5. 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')
6. 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消息格式(好比线上环境，修改比较困难)，可以重新实现

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

复制代码

4.3、自界说Kafka消息解析器

        生产中对Kafka消息及解析的格式总是各种各样的，当flink预界说的解析器满意不了业务需求时，可以通过自界说kafka消息解析器来完成业务的支持
比方，当利用 MyJSONKeyValueDeserializationSchema 获取Kafka元数据时，只返回了 offset、topic、partition 三个字段信息，现在需要`kafka生产者写入数据时的timestamp`，就可以通过自界说kafka消息解析器来完成
代码示例：

1. // TODO 自定义Kafka消息解析器，在 metadata 中增加 timestamp字段
2. public class MyJSONKeyValueDeserializationSchema implements KafkaDeserializationSchema<ObjectNode>{
4.         private static final long serialVersionUID = 1509391548173891955L;
6.         private final boolean includeMetadata;
7.         private ObjectMapper mapper;
9.         public MyJSONKeyValueDeserializationSchema(boolean includeMetadata) {
10.             this.includeMetadata = includeMetadata;
11.         }
13.         @Override
14.         public void open(DeserializationSchema.InitializationContext context) throws Exception {
15.             mapper = JacksonMapperFactory.createObjectMapper();
16.         }
18.         @Override
19.         public ObjectNode deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
20.             ObjectNode node = mapper.createObjectNode();
21.             if (record.key() != null) {
22.                 node.set("key", mapper.readValue(record.key(), JsonNode.class));
23.             }
24.             if (record.value() != null) {
25.                 node.set("value", mapper.readValue(record.value(), JsonNode.class));
26.             }
27.             if (includeMetadata) {
28.                 node.putObject("metadata")
29.                         .put("offset", record.offset())
30.                         .put("topic", record.topic())
31.                         .put("partition", record.partition())
32.                         // 添加 timestamp 字段
33.                         .put("timestamp",record.timestamp())
34.                 ;
35.             }
36.             return node;
37.         }
39.         @Override
40.         public boolean isEndOfStream(ObjectNode nextElement) {
41.             return false;
42.         }
44.         @Override
45.         public TypeInformation<ObjectNode> getProducedType() {
46.             return getForClass(ObjectNode.class);
47.         }
48.     }

复制代码

运行效果：

---

5、起始消费位点应该怎样设置

起始消费位点说明：
        起始消费位点是指 启动flink使命时，应该从哪个位置开始读取Kafka的消息   
        下面先容下常用的三个设置：    
                OffsetsInitializer.earliest()  ：
                        从最早位点开始消
                        这里的最早指的是Kafka消息生存的时长(默以为7天，天生环境各公司略有不同)
                        该这设置为默认设置，当不指定OffsetsInitializer.xxx时，默以为earliest() 
                OffsetsInitializer.latest()   ：
                        从最末了位点开始消费
                        这里的最末了指的是flink使命启动时间点之后生产的消息
                OffsetsInitializer.timestamp(时间戳) ：
                        从时间戳大于等于指定时间戳（毫秒）的数据开始消费
下面用案例说明下，三种设置的效果，kafak天生10条数据，如下：

5.1、earliest()

代码示例：

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

复制代码

运行效果：

5.2、latest()

代码示例：

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

复制代码

运行效果：

5.3、timestamp()

代码示例：

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

复制代码

运行效果：

---

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

        在flink1.13的时候，如果Kafka分区扩容了，只有通过重启flink使命，才气消费到新增分区的数据，小编就曾遇到过上游业务部门的kafka分区扩容了，并没有关照卑鄙利用方，导致及时指标异常，以致丢失了数据。
        在flink1.17的时候，可以通过`开启动态分区查抄`，来实现不用重启flink使命，就能消费到新增分区的数据
开启分区查抄：(默认不开启)

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

复制代码

代码示例：

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

复制代码

---

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

可以在 fromSource(source,WatermarkStrategy,sourceName) 时，提取事件时间和制定水位线天生战略
注意：当不指定事件时间提取器时，Kafka Source 利用 Kafka 消息中的时间戳作为事件时间
7.1、利用内置的单调递增的水位线天生器 + kafka timestamp 为事件时间

代码示例：

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

复制代码

运行效果：

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

代码示例：

1.     // 在读取Kafka消息时，提取事件时间&插入水位线
2.     public static void KafkaSourceExtractEventtimeAndWatermark() throws Exception {
3.         // 1.获取执行环境
4.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
6.         // 2.读取kafka数据
7.         KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder()
8.                 .setBootstrapServers("worker01:9092")
9.                 .setTopics("9527")
10.                 .setGroupId("FlinkConsumer")
11.                 // 将kafka消息解析为Json对象，并返回元数据
12.                 .setDeserializer(KafkaRecordDeserializationSchema.of(
13.                         new MyJSONKeyValueDeserializationSchema(true)
14.                 ))
15.                 // 设置起始消费位点：从最末尾位点开始消费
16.                 .setStartingOffsets(OffsetsInitializer.latest())
17.                 .build();
19.         env.fromSource(source,
20.                         // 使用内置的单调递增的水位线生成器（使用 kafka消息中的ID字段作为事件时间）
21.                         WatermarkStrategy.<ObjectNode>forMonotonousTimestamps()
22.                                 // 提取 Kafka消息中的 ID字段作为 事件时间
23.                                 .withTimestampAssigner(
24.                                         (json, timestamp) -> Long.parseLong(json.get("value").get("ID").toString())
25.                                 ),
27.                         "Kafka Source")
28.                 // 通过 ProcessFunction 查看提取的事件时间和水位线信息
29.                 .process(
30.                         new ProcessFunction<ObjectNode, String>() {
31.                             @Override
32.                             public void processElement(ObjectNode kafkaJson, ProcessFunction<ObjectNode, String>.Context ctx, Collector<String> out) throws Exception {
33.                                 // 当前处理时间
34.                                 long currentProcessingTime = ctx.timerService().currentProcessingTime();
35.                                 // 当前水位线
36.                                 long currentWatermark = ctx.timerService().currentWatermark();
37.                                 StringBuffer record = new StringBuffer();
38.                                 record.append("========================================\n");
39.                                 record.append(kafkaJson + "\n");
40.                                 record.append("currentProcessingTime：" + currentProcessingTime + "\n");
41.                                 record.append("currentWatermark：" + currentWatermark + "\n");
42.                                 record.append("kafka-ID：" + Long.parseLong(kafkaJson.get("value").get("ID").toString()) + "\n");
43.                                 record.append("kafka-timestamp：" + Long.parseLong(kafkaJson.get("metadata").get("timestamp").toString()) + "\n");
44.                                 out.collect(record.toString());
46.                             }
47.                         }
48.                 ).print();
50.         // 3.触发程序执行
51.         env.execute();
52.     }

复制代码

运行效果：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。