flinkkafkaconsumer（flinkkafkaconsumer参数优化）

by intanet.cn ca 大数据 on 2024-03-18

本篇文章给大家谈谈flinkkafkaconsumer，以及flinkkafkaconsumer参数优化对应的知识点，希望对各位有所帮助，不要忘了收藏本站喔。

本文目录一览：

1、Flink是如何从kafka中拉取数据的
2、Flink如何管理Kafka 消费位点(译文)
3、Flink 配置Kafka数据源
4、flink与kafka结合
5、FlinkKafkaConsumer官网文档翻译
6、flink消费kafka细节

Flink是如何从kafka中拉取数据的

首先来看一下 FlinkKafkaConsumerBase.run方法，相当于是Flink 从kafka中拉取数据的入口方法:

createFetcher方法

返回了一个亮链 KafkaFetcher对象，我们点进去看一下

KafkaFetcher的构造器里面创建了一个 KafkaConsumerThread对象

至此为止createFetch就介绍完了，也可敬轿孙以看作是拉取数据的准备工作，接下来看一下kafkaFetcher.runFetchLoop();

KafkaFetch中的runFetchLoop方法，正式开始从kafka中拉取message

既然consumerThread.start()开始了实际的kafka consumer，我们一起来看一下consumerThread中的方法

至此如何从kafka中拉取帆旁数据，已经介绍完了

Flink如何管理Kafka 消费位点(译文)

Checkpointing 是 Flink 故障恢复的内部机制。一个 checkpoint 就是 Flink应用程序产生的状态的一个副本。如果 Flink 任务发生故障，它会从 checkpoint 中载入之前的状态来恢复任务，就好像故障没有发生一样。

Checkpoints 是 Flink 容错的基础，并且确保了 Flink 流式应用在失败时的完整性。Checkpoints 可以通过 Flink 设置定时触发。

Flink Kafka consumer 使用 Flink 的 checkpoint 机制来存储 Kafka 每个分区的位点到 state。当 Flink 执行 checkpoint 时，Kafka 的每个分区的位点都被存储到 checkpoint 指定的 filesystem 中。Flink 的 checkpoint 机制确保了所有任务算子的状态是一致的，也就是说这些状态具有相同的数据输入。当所有的任务算子成功存储他们自己的状态后，代表一次 checkpoint 的完成。因此，当任务从故障中恢复时，Flink 保证了exactly-once。

下面将一步一步的演示 Flink 是如何通过 checkpoint 来管理 Kafka 的 offset 的。

下面的例子从两个分区的 Kafka topic 中读取数据，每个分区的数据是 “A”, “B”, “C”, ”D”, “E”。假设每个分区都是从 0 开始读取。

假设 Flink Kafka consumer 从分区 0 开始读取数据 “A”，那么此时第一个 consumer 的位点从 0 变成 1。如下图所示。

此时数据 “A” 到达 Flink Job 中的 Map Task。两个 consumer 继续读取数据 (从分区 0 读取数据 “B” ，从分区 1 读取数据 “A”)。 offsets 分别被更新成 2 和 1。与此同时，假设 Flink 从 source 端开始执行 checkpoint。

到这里，Flink Kafka consumer tasks 已经执行了一次快照，offsets也保存到了 state 中(“offset = 2, 1”) 。此时 source tasks 在数据 “B” 和 “A” 后面，向下游发送一个 checkpoint barrier。checkpoint barriers 是 Flink 用来对齐每个任务算子的 checkpoint，以确保整个 checkpoint 的一致性。分区 1 的数据 “A” 到达 Flink Map Task，与此同时分区 0 的 consumer 继续读取下一首迹个消息(message “C”)。

Flink Map Task 收到上游两个 source 的 checkpoint barriers 然后开始执行 checkpoint ，把 state 保存到 filesystem。同时，消费者继续从Kafka分区读取更多事件。

假设 Flink Map Task 是 Flink Job 的最末端，那么当它完成 checkpoint 后，就会立马通知 Flink Job Master。当 job 的所有 task 都确认其 state 已经 “checkpointed”，Job Master将完成这次的整个 checkpoint。之者困并后，checkpoint 可以用于故障恢复。

如果发生故障（例如，worker 挂掉），则所有任务将重启，并且它们的状态将被重置为最近一次的 checkpoint 的状态。如下图所示尺坦。

source 任务将分别从 offset 2 和 1 开始消费。当任务重启完成，将会正常运行，就像之前没发生故障一样。

PS: 文中提到的 checkpoint 对齐，我说下我的理解，假设一个 Flink Job 有 Source - Map - Sink，其中Sink有多个输入。那么当一次checkpoint的 barrier从source发出时，到sink这里，多个输入需要等待其它的输入的barrier已经到达，经过对齐后，sink才会继续处理消息。这里就是exactly-once和at-least-once的区别。

The End

原文链接: How Apache Flink manages Kafka consumer offsets

Flink 配置Kafka数据源

flink 中已经预置了 kafka 相关的数据源实现 FlinkKafkaConsumer010 ，先看下具体的实现：

kafka 的 Consumer 有一堆实现，不过最终都是继承自 FlinkKafkaConsumerBase ，而这个抽象类则是继承 RichParallelSourceFunction ，是不是很眼熟，跟自定义 mysql 数据源继承的抽象类 RichSourceFunction 很类似液空。

可以看到，这里有很多构造函数，我们直接使用即可。

说明：铅岩

a、这里直接使用 properties 对象来设置 kafka 相关配置，比如 brokers 、 zk 、 groupId 、序列化、反序列化等。

b、使用 FlinkKafkaConsumer010 构造函数，指定 topic 、 properties 配置

c、 SimpleStringSchema 仅针对 String 类型数据的序列化及反序列化，如果 kafka 中消息的内容不是 String ，则会报错；看下 SimpleStringSchema 的定义：

d、这里直接把获取槐埋御到的消息打印出来。

[img]

flink与kafka结合

flink提供了一个瞎铅禅特有的kafka connector去读写kafka topic的数据。flink消费kafka数据，并不是完全通过跟踪kafka消费组的offset来实现去保证exactly-once的语义，而是flink内部去跟踪offset和做checkpoint去实现exactly-once的语义

flink与kafka整合，相应版本对于的maven依赖如下表

maven依赖举例

flink.version1.7.0/flink.version

scala.binary.version2.11/scala.binary.version

scala.version2.11.12/scala.version

dependency

groupIdorg.apache.flink/groupId

artifactIdflink-streaming-scala_${scala.binary.version}/artifactId

version${flink.version}/version

scopeprovided/scope

/dependency

flink利用FlinkKafkaConsumer来读取访问kafka, 根据kafka版本不同FlinkKafkaConsumer的类名也会变化，会变为FlinkKafkaConsumer

[08,09,10...]后面的数字就是对于的kafka的大版本号。

初始化FlinkKafkaConsumer 需要如下参数

1、topic名字，用来指定消费一个或者多个topic的数据

2、kafka的配置信息，如zk地址端激销口，kafka地址端口等

3、反序列化器(schema)，对消费数据选择一个反序列化器进行反序列化。

flink kafka的消磨尘费端需要知道怎么把kafka中消息数据反序列化成java或者scala中的对象。用户通过使用DeserializationSchema，每一条kafka的消息都会作用于DeserializationSchema的eserialize(byte[] message)方法。来将kafka的消息转换成用户想要的结构。

用户通过自定义schema将接入数据转换成自定义的数据结构，主要通过实现KeyedDeserializationSchema或者DeserializationSchema接口来完成，可以自定义。flink内置的对DeserializationSchema 的实现有

public class SimpleStringSchema implements DeserializationSchemaString

public class TypeInformationSerializationSchemaT implements DeserializationSchemaT

对 KeyedDeserializationSchema的实现有

public class TypeInformationKeyValueSerializationSchemaK, V implements KeyedDeserializationSchemaTuple2K, V

public class JSONKeyValueDeserializationSchema implements KeyedDeserializationSchemaObjectNode

例如：

val myConsumer = new FlinkKafkaConsumer010[String]("topic",new SimpleStringSchema,p)

public class MySchema implements KeyedDeserializationSchemaKafkaMsgDTO {

@Override

public KafkaMsgDTO deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {

String msg = new String(message, StandardCharsets.UTF_8);

String key = null;

if(messageKey != null){

key = new String(messageKey, StandardCharsets.UTF_8);

}

return new KafkaMsgDTO(msg,key,topic,partition,offset);

}

@Override

public boolean isEndOfStream(KafkaMsgDTO nextElement) {

return false;

}

@Override

public TypeInformationKafkaMsgDTO getProducedType() {

return getForClass(KafkaMsgDTO.class);

}

dependency

groupIdorg.apache.flink/groupId

artifactIdflink-connector-kafka-base_2.11/artifactId

version1.7.0/version

/dependency

public class KafkaMsgDTO {

private String topic;

private int partition;

private long offset;

private String mesg;

@Override

public String toString() {

return "KafkaMsgDTO{" +

"topic='" + topic + '\'' +

", partition=" + partition +

", offset=" + offset +

", mesg='" + mesg + '\'' +

", key='" + key + '\'' +

'}';

}

private String key;

public KafkaMsgDTO(){

}

public KafkaMsgDTO(String mesg,String key,String topic,int partition,long offset){

this.mesg = mesg;

this.key = key;

this.topic = topic;

this.partition = partition;

this.offset = offset;

}

public String getKey() {

return key;

}

public void setKey(String key) {

this.key = key;

}

public String getTopic() {

return topic;

}

public void setTopic(String topic) {

this.topic = topic;

}

public int getPartition() {

return partition;

}

public void setPartition(int partition) {

this.partition = partition;

}

public long getOffset() {

return offset;

}

public void setOffset(long offset) {

this.offset = offset;

}

public String getMesg() {

return mesg;

}

public void setMesg(String mesg) {

this.mesg = mesg;

}

val myConsumer = new FlinkKafkaConsumer010[KafkaMsgDTO]("topic",new MySchema(),p)

// myConsumer.setStartFromEarliest()

//从最早开始消费，消费过的数据会重复消费，从kafka来看默认不提交offset.

// myConsumer.setStartFromLatest()

//从最新开始消费，不消费流启动前未消费的数据，从kafka来看默认不提交offset.

myConsumer.setStartFromGroupOffsets()

//从消费的offset位置开始消费，kafka有提交offset，这是默认消费方式

//如果没有做checkpoint 数据进入sink就会提交offset,如果sink里面逻辑失败。offset照样会提交，程序退出，如果重启流，消费失败的数据不会被重新消费

//如果做了checkpoint 会保证数据的端到端精准一次消费。sink里面逻辑失败不会提交offset

env.enableCheckpointing(5000);

val stream = env.addSource(myConsumer)

stream.addSink(x={

println(x)

println(1/(x.getMesg.toInt%2))//消息是偶数就会报错，分母为0

println(x)

})

val stream = env.addSource(myConsumer)

//实验表明如果sink处理逻辑有一部线程在跑，如果异步线程失败。offset照样会提交。

stream.addSink(x={

println(x)

new Thread(new Runnable {

override def run(): Unit = {

println(1/(x.getMesg.toInt%2))//消息是偶数就会报错，分母为0

}

}).start()

println(x)

})

val specificStartOffsets = new java.util.HashMap[KafkaTopicPartition, java.lang.Long]()

specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L)

specificStartOffsets.put(new KafkaTopicPartition("myTopic", 1), 31L)

specificStartOffsets.put(new KafkaTopicPartition("myTopic", 2), 43L)

myConsumer.setStartFromSpecificOffsets(specificStartOffsets)

FlinkKafkaConsumer官网文档翻译

以下内容翻译者段空自 Apache Flink Kafka Connector官网(内容顺序稍作修改)

Flink为Kafka topic读取和写入数据提供了特殊的Kafka连接器。Flink Kafka消费者与Flink的检查点机制集成可以保证下游的exactly-once语义。为了实现这一点，Flink并不完全依赖Kafka自身维护的消费者组offset，而燃渣是在Flink内部管理这些offset。

通用kafka（1.0.0版本及以后）：

kafka版本（0.11版本及以前）：

构造器有以下参数：

Flink Kafka Consumer如何将Kafka中的二进制数据转换为Java / Scala对象。这就需要指定序列化和反序列化方式。每个Kafka消息都会调用 T deserialize(byte[] message) 方法。

后边的太难了，自己去官网看吧：

说明：

启用Flink的checkpoint机制后，Flink Kafka Consumer在消费kafka消息的同时，会周期的并保持一致性的将offset写入checkpoint中。如果作业失败，Flink会将程序恢复到最新的checkpoint的状态，并从存储在checkpoint中的偏移量开始重新消费Kafka中的消息。

注意，只有在有足够的slots可用来首瞎重新启动时，Flink才能重新启动。

如果没有启用checkpoint机制，Kafka使用者将定期向Zookeeper提交偏移量。

Flink Kafka Consumer可以将偏移量提交到Kafka broker(或0.8中的Zookeeper)。注意，Flink Kafka Consumer并不依赖提交的偏移量来保证容错。提交的偏移量只是用于方便查看Consumer消费的进度。

提交偏移量有多种不同的方式，与是否为Job启用了checkpoint机制有关。

flink消费kafka细节

# flink消费kafka细节

Apache kafka connector提供对Kafka服务的事件流的访问。Flink提供了特殊的Kafka连接器，用于从Kafka主题读写数据。 Flink Kafka Consumer与Flink的检查点机制集成在一起，以提供一次精确的处理语义。为此，Flink不仅仅依赖于Kafka的消费者群体偏移量跟踪，还内部跟踪和检查这些好樱偏移量。

请为您的用例和环境选择一厅袜含个包（Maven项目ID）和类名扮笑。对于大多数用户来说，FlinkKafkaConsumer08（flink-connector-kafka的一部分）是合适的。

| :------------------------------ | :-------------- | :------------------------------------------ | :------------ | :----------------------------------------------------------- |

| flink-connector-kafka-0.11_2.11 | 1.4.0 | FlinkKafkaConsumer011 FlinkKafkaProducer011 | 0.11.x | Since 0.11.x Kafka does not support scala 2.10. This connector supports [Kafka transactional messaging]() to provide exactly once semantic for the producer. |

| flink-connector-kafka_2.11 | 1.7.0 | FlinkKafkaConsumer FlinkKafkaProducer | = 1.0.0 | This universal Kafka connector attempts to track the latest version of the Kafka client. The version of the client it uses may change between Flink releases. Starting with Flink 1.9 release, it uses the Kafka 2.2.0 client. Modern Kafka clients are backwards compatible with broker versions 0.10.0 or later. However for Kafka 0.11.x and 0.10.x versions, we recommend using dedicated flink-connector-kafka-0.11_2.11 and flink-connector-kafka-0.10_2.11 respectively. |

在创建kafka consumer时，需要指定一些参数

```java

Properties properties = new Properties();

// kafka broker地址，用逗号隔开

properties.setProperty("bootstrap.servers", "localhost:9092");

// zookeeper机器地址，仅在Kafka 0.8用到

properties.setProperty("zookeeper.connect", "localhost:2181");

// kafka消费者的group.id

properties.setProperty("group.id", "test");

DataStreamString stream = env

.addSource(new FlinkKafkaConsumer08("topic", new SimpleStringSchema(), properties));

```

### flink消费kafka时的容错

启用flink检查点之后，flink会定期checkpoint offset，万一作业失败，Flink将把流式程序恢复到最新检查点的状态，并从存储在检查点的偏移量开始重新使用Kafka的记录。

```java

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

env.enableCheckpointing(5000); // checkpoint every 5000 msecs

```

### flink worker和kafka partition对应关系

partition会分配给flink并行的task，当task比partition数量多时，会有task进程闲置

![Fewer Partitions](./fewer_partition.png)

当kafka的partition比flink task多时，一个task会分配到多个partition

![More Partitions](./more_partition.png)

### flink如何保证kafka的恰好一次处理

flink kafka consumer和flink的检查点机制紧密集成，flink每次从kafka队列读到新数据都会更新offset到state中，flink kafka consumer是一个stateful operator，其state存的就是offset。

### 从Kafka主题阅读时，Flink如何处理背压？

如果下游无法以相同的速度处理所有传入数据，则像Flink这样的流媒体系统必须能够减慢flink kafka consumer消费的速度。这就是所谓的反压处理。 Flink的Kafka consumer自带处理backpressure的机制：一旦下游无法跟上Kafka消费的速度，Flink就会放慢来自Kafka的消息的消费，从而减少来自代理的请求。由于代理将所有消息都保留在磁盘上，因此它们也可以提供过去的消息。一旦操作员再次加速，Flink将全速消耗累积的消息。这种行为使Kafka非常适合作为流源和Flink之间的缓冲区，因为它为负载高峰时的事件提供了持久的缓冲区。

### kafka生产者API的使用

关于flinkkafkaconsumer和flinkkafkaconsumer参数优化的介绍到此就结束了，不知道你从中找到你需要的信息了吗？如果你还想了解更多这方面的信息，记得收藏关注本站。

opencv库（opencv库安装不上）包含idea大写的词条