flinkkafkaconsumer（flinkkafkaconsumer数据来源的topic）

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

本文目录一览：

• 1、flink消费kafka细节
• 2、flink与kafka结合
• 3、FlinkKafkaConsumer010 代码阅读笔记

flink消费kafka细节

# flink消费kafka细节

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

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

| Maven Dependency                | Supported since | Consumer and Producer Class name            | Kafka version | Notes                                                        |

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

| flink-connector-kafka-0.8_2.11  | 1.0.0          | FlinkKafkaConsumer08 FlinkKafkaProducer08  | 0.8.x        | Uses the [SimpleConsumer]() API of Kafka internally. Offsets are committed to ZK by Flink. |

| flink-connector-kafka-0.9_2.11  | 1.0.0          | FlinkKafkaConsumer09 FlinkKafkaProducer09  | 0.9.x        | Uses the new [Consumer API]() Kafka. |

| flink-connector-kafka-0.10_2.11 | 1.2.0          | FlinkKafkaConsumer010 FlinkKafkaProducer010 | 0.10.x        | This connector supports [Kafka messages with timestamps]() both for producing and consuming. |

| 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进程闲置


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


### 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的使用

[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)

FlinkKafkaConsumer010 代码阅读笔记

FlinkKafkaConsumer010 是 flink 1.6.1 提供的 Kafka 数运手尘据源接入实现，在 flink 框架中数据源需要实现 SourceFunction 接口。

SourceFunction 声明了两个接口方法：

flink 通过 SourceContext 提供 element 输出的接口：

至于各种 Time 与 Watermark 之间的关系可以参考 官方文档 ，这里就不赘述了。

首薯宴先通过类图对 FlinkKafkaConsumer010 进行整体的认识。

ParallelSourceFunction 是 SourceFunction 的子类，实际上该类是一个标签，用于通知系统该 source 可以并行执行。

RichFunction 提供了 open 和 close 两个钩子方法，用于开始前和结束后回调执行；另外还提供了 RuntimeContext 的设置获取方法。

先来看 FlinkKafkaConsumer010 的构建逻辑，重点在 FlinkKafkaConsumer09 中：

顺着看父类 FlinkKafkaConsumerBase 的构造方法：

初始化的流程主要包含两部分：CheckpointedFunction.initializeState 和RichFunction.open。

在FlinkKafkaConsumerBase 中：

从检查点中恢复 offset 到 restoredState 中。

其中对于 1.2 版本的状态也做了兼容，这里的细节就不看了。

在FlinkKafkaConsumerBase 中：

回想一下，这个是 ParallelSourceFunction，也就是说会有多个实例并行执行，所以不同的实例需要分配一下 TopicPartition。

如果从检查点恢复了状态，那么依据检查点的内容设置 subscribedPartitionsToStartOffsets，对于新增的 partition 设置为 EARLIEST_OFFSET；

如果没有从检查点恢复状态，那么根据 startupMode 设置 subscribedPartitionsToStartOffsets。

startupMode 默认为 GROUP_OFFSETS，也就是依据 Kafka 消费组提交的 offset 继续消费。

执行逻辑的接口方法是 SourceFunction 的 run。

具体实现在 FlinkKafkaConsumerBase 中：

通过抽象方法 createFetcher 方法创建 AbstractFetcher，具体的实现在子类 FlinkKafkaConsumer010 中。

关键有三个模块：unassignedPartitionsQueue、Handover 和 consumerThread。

这是一个 ClosableBlockingQueueKafkaTopicPartitionStateKPH 队列，当初始化时会把需要消旁禅费的 TopicPartition 加入这个队列；如果启动了 TopicPartition 周期性自动发现，那么后续新发现 TopicPartition 也会加入这个队列。

可以理解为一个长度为一的阻塞队列，将 consumerThread 获取的消息或者抛出的异常，传递给 flink 执行的线程。

执行 runFetchLoop 方法

从 handover 中获取 ConsumerRecords，经过 deserializer 反序列化后 emit 出去。

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