步骤

实验一：

linux:

win:

结果：

实验二：

linux

win

结果：

代码

KafkaEOSDemo

KafkaEOSDemo2

pom.xml

文档

KafkaEOSDemo

1. 创建环境

2. 启用检查点

3. 设置检查点存储路径

4. 从Kafka读取数据

1. 调用setBootstrapServers(hadoop102:9092,hadoop103:9092,hadoop104:9092)设置Kafka的服务器地址，也就是Kafka broker的地址。

2. 调用setGroupId(test)设置消费者组ID，Kafka使用消费者组来区分不同的消费者。

3. 调用setTopics(topic_1)设置要订阅的Kafka主题。

4. 调用setValueOnlyDeserializer(new SimpleStringSchema())设置反序列化器，这里使用的是简单字符串反序列化器。

5. 调用setStartingOffsets(OffsetsInitializer.latest())设置起始偏移量，这里设置为从最新的数据开始消费。

6. 调用build()构建KafkaSource实例。

5. 将数据写入到另一个Kafka主题

1. 同KafkaSource的setBootstrapServers。

2. setTopic(ws)：设置要写入的Kafka主题。

3. setValueSerializationSchema(new SimpleStringSchema())：设置序列化器，这里我们使用的是简单字符串序列化器。

4. setRecordSerializer(...)：设置记录序列化器，用于将Flink的数据记录序列化为Kafka可以接受的格式。

5. setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)：设置交付保证，这里我们设置为精确一次，也就是说每条记录将被精确地写入一次。

6. setTransactionalIdPrefix(prefix-)：设置事务ID前缀，这是用于Kafka事务的标识。

7. setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + )：设置事务超时时间，这是Kafka事务的一个重要参数，它决定了一个事务可以保持打开状态的最长时间。间。

8. build()：构建KafkaSink实例。

KafkaEOSDemo2

1. 创建环境

2. 从Kafka读取数据

1. setBootstrapServers("hadoop102:9092,hadoop103:9092,hadoop104:9092")：设置Kafka的服务器地址，也就是Kafka broker的地址。

2. setGroupId("test")：设置消费者组ID，Kafka使用消费者组来区分不同的消费者。

3. setTopics("ws")：设置要订阅的Kafka主题。

4. setValueOnlyDeserializer(new SimpleStringSchema())：设置反序列化器，这里我们使用的是简单字符串反序列化器。

5. setStartingOffsets(OffsetsInitializer.latest())：设置起始偏移量，这里我们设置为从最新的数据开始消费。

6. setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")：设置事务的隔离级别为读已提交，这是Kafka事务的一个重要配置。读已提交的隔离级别表示消费者在读取数据时，只能读取到已经被提交的事务。

7. build()：构建KafkaSource实例。