Flink实战（八）Flink 使用 Kafka Source & Kafka Sink

比眉伴天荒 2023-06-09 10:23 50阅读 0赞

# FlinkKafkaConnector #

该连接器提供对Apache Kafka服务的事件流的访问。

Flink提供了特殊的Kafka连接器，用于从Kafka主题读写数据。 Flink Kafka Consumer与Flink的检查点机制集成在一起，以提供**有且仅有一次**的语义。为此，Flink不仅仅依赖于Kafka的消费者群体偏移量跟踪，还内部跟踪和检查这些偏移量。

## 开发流程 ##

接一下以一个示例配置来介绍一下`Flink`如何以读取Kafka，并且再写入到一个新的Topic中

1.读取

2.写入

2.写入终端

KafkaSource topic\_demo

Flink

KafkaSource topic\_demo\_flink

标准输出

### 1. 依赖kafka ###

pom.xml 添加依赖

<dependency>
    		<groupId>org.apache.flink</groupId>
    		<artifactId>flink-connector-kafka_2.11</artifactId>
    		<version>1.9.0</version>
    	</dependency>

### 2. 连接Kafka&写入Kafka ###

public static final String BOOTSTRAP_SERVERS = "192.168.121.101:9092";
    
      public static void main(String[] args) throws Exception { 
    
          StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
          // 1.0 配置KafkaConsumer
          Properties props = new Properties();
          props.setProperty("bootstrap.servers", BOOTSTRAP_SERVERS);
          props.setProperty("group.id", "test111");
          props.put("enable.auto.commit", "true");
          props.put("auto.commit.interval.ms", "1000");
          props.put("session.timeout.ms", "30000");
          props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
          // 1.1 把kafka设置为source
          env.enableCheckpointing(5000); // checkpoint every 5000 msecs
          DataStream<String> stream = env
                  .addSource(new FlinkKafkaConsumer<>("demo", new SimpleStringSchema(), props));
    
          // 2.0 配置 kafkaProducer
          FlinkKafkaProducer<String> myProducer = new FlinkKafkaProducer<String>(
                  BOOTSTRAP_SERVERS,            // broker list
                  "demo_flink",                  // target topic
                  new SimpleStringSchema());   // serialization schema
          myProducer.setWriteTimestampToKafka(true);
    
          // 2.1 把kafka设置为sink
          stream.addSink(myProducer);
          
          // 2.2 把kafka接收debug打印出来
          stream.print();
    
          env.execute("Kafka Source");
      }

--------------------

## 容错保证 ##

在真实的生产环境上，我们都需要保证系统的高可用。即需要保证系统的各个组件不能出现问题，或者提供一系列的容错机制。

### Kafka 消费者的容错 ###

启用Flink的检查点后，Flink Kafka Consumer将在一个topic消费记录的时候，并以一致的方式定期记录Kafka偏移量和其它操作者的操作到检查点。万一作业失败，Flink将把流式程序恢复到最新检查点的状态，并从检查点中存储的偏移量开始重新使用Kafka的记录。

要使用容错的Kafka使用者，需要在执行环境中启用拓扑检查点：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.enableCheckpointing(5000); // checkpoint every 5000 msecs

如果未启用检查点，则Kafka使用者将定期将偏移量提交给`Zookeeper`。

### Kafka 生产者的容错 ###

#### Kafka 0.8 ####

在0.9之前Kafka没有提供任何机制去保证至少一次和仅仅一次的语义

#### Kafka 0.9 and 0.10 ####

启用Flink的检查点后，FlinkKafkaProducer09和FlinkKafkaProducer010可以提供`至少一次`的保证。

除了启用Flink的检查点之外，您还应该适当配置设置方法setLogFailuresOnly（boolean）和setFlushOnCheckpoint（boolean）。

*  `setLogFailuresOnly(boolean)`: 默认情况下，它设置为false。启用此功能将使生产者仅记录故障，而不是捕获并重新抛出故障。从本质上讲，即使从未将记录写入目标Kafka主题，该记录也取得了成功。如果需要`至少一次`必须禁用此功能。
 *  `setFlushOnCheckpoint(boolean)`: 默认情况下，此设置为true。启用此功能后，Flink的检查点将在Kafka确认该检查点时等待所有即时记录，然后再执行该检查点。这样可以确保将检查点之前的所有记录都写入Kafka。必须至少启用一次。如果需要`至少一次`必须启用此功能。

总之，默认情况下，Kafka生产者对版本0.9和0.10`至少具有一次`语义的保证，其中`setLogFailureOnly`设置为false，`setFlushOnCheckpoint`设置为true。

#### Kafka 0.11 and newer ####

启用Flink的检查点后，FlinkKafkaProducer011（适用于Kafka> = 1.0.0版本的FlinkKafkaProducer）可以提供`有且仅有一次`语义的保证。

除了启用Flink的检查点之外，您还可以通过将适当的语义参数传递给FlinkKafkaProducer011（适用于Kafka> = 1.0.0版本的FlinkKafkaProducer）来选择三种不同的操作模式：

*  `Semantic.NONE`: Flink不会保证任何是，生产的消息可能会丢失或者重复发送
 *  `Semantic.AT_LEAST_ONCE` (默认配置): 类似于 `setFlushOnCheckpoint(true)` 在`FlinkKafkaProducer010`. 可以保证消息不会丢失，但是不保证会重复发送
 *  `Semantic.EXACTLY_ONCE`: 使用Kafka事务提供`有且仅一次`的语义.每当您使用事务写入Kafka时，请不要忘记设置所需的`隔离级别` (`read_committed` or `read_uncommitted` \- 后一个是默认值) 对于使用Kafka记录的任何应用程序。