【Flink1.9整合Kafka】
文章图片

本文基于Flink1.9版本简述如何连接Kafka。
流式连接器
文章图片

我们知道可以自己来开发Source 和 Sink ，但是一些比较基本的 Source 和 Sink 已经内置在 Flink 里。
预定义的source支持从文件、目录、socket，以及 collections 和 iterators 中读取数据。
预定义的sink支持把数据写入文件、标准输出（stdout）、标准错误输出（stderr）和 socket。
连接器可以和多种多样的第三方系统进行交互。目前支持以下系统:

• Apache Kafka
• Apache Cassandra(sink)
• Amazon Kinesis Streams(source/sink)
• Elasticsearch(sink)
• Hadoop FileSystem (sink)
• RabbitMQ(source/sink)
• Apache NiFi(source/sink)
• Twitter Streaming API(source)

请记住，在使用一种连接器时，通常需要额外的第三方组件，比如：数据存储服务器或者消息队列。
Apache Bahir 中定义了其他一些连接器

• Apache ActiveMQ(source/sink)
• Apache Flume(sink)
• Redis(sink)
• Akka (sink)
• Netty (source)

使用connector并不是唯一可以使数据进入或者流出Flink的方式。一种常见的模式是从外部数据库或者 Web 服务查询数据得到初始数据流，然后通过 Map 或者 FlatMap 对初始数据流进行丰富和增强，这里要使用Flink的异步IO。
而向外部存储推送大量数据时会导致 I/O 瓶颈问题出现。在这种场景下，如果对数据的读操作远少于写操作，可以让外部应用从 Flink 拉取所需的数据，需要用到Flink的可查询状态接口。
本文重点介绍Apache Kafka Connector
Kafka连接器 此连接器提供对Apache Kafka提供的事件流的访问。
Flink提供特殊的Kafka连接器，用于从/向Kafka主题读取和写入数据。Flink Kafka Consumer集成了Flink的检查点机制，可提供一次性处理语义。为实现这一目标，Flink并不完全依赖Kafka 的消费者组的偏移量，而是在内部跟踪和检查这些偏移。
下表为不同版本的kafka与Flink Kafka Consumer的对应关系。

Maven Dependency	Supported since	Consumer and Producer Class name	Kafka version
flink-connector-kafka-0.8_2.11	1.0.0	FlinkKafkaConsumer08 FlinkKafkaProducer08	0.8.x
flink-connector-kafka-0.9_2.11	1.0.0	FlinkKafkaConsumer09 FlinkKafkaProducer09	0.9.x
flink-connector-kafka-0.10_2.11	1.2.0	FlinkKafkaConsumer010 FlinkKafkaProducer010	0.10.x
flink-connector-kafka-0.11_2.11	1.4.0	FlinkKafkaConsumer011 FlinkKafkaProducer011	0.11.x
flink-connector-kafka_2.11	1.7.0	FlinkKafkaConsumer FlinkKafkaProducer	>= 1.0.0

而从最新的Flink1.9.0版本开始，使用Kafka 2.2.0客户端。
下面简述使用步骤。
导入maven依赖：

org.apache.flink flink-connector-kafka_2.11 1.9.0

安装Kafka：
可以参照 Kafka入门宝典（详细截图版）
兼容性：
从Flink 1.7开始，它不跟踪特定的Kafka主要版本。相反，它在Flink发布时跟踪最新版本的Kafka。如果您的Kafka代理版本是1.0.0或更高版本，则应使用此Kafka连接器。如果使用旧版本的Kafka（0.11,0.10,0.9或0.8），则应使用与代理版本对应的连接器。
升级Connect要注意Flink升级作业，同时

• 在整个过程中使用Flink 1.9或更新版本。
• 不要同时升级Flink和运营商。
  
• 确保您作业中使用的Kafka Consumer和/或Kafka Producer分配了唯一标识符（uid）。
  
• 使用stop with savepoint功能获取保存点（例如，使用stop --withSavepoint）。
  

用法：
引入依赖后，实例化新的source（FlinkKafkaConsumer）和sink（FlinkKafkaProducer）。
Kafka Consumer 先分步骤介绍构建过程，文末附Flink1.9连接Kafka完整代码。
Kafka consumer 根据版本分别叫做FlinkKafkaConsumer08 FlinkKafkaConsumer09等等
Kafka >= 1.0.0 的版本就叫FlinkKafkaConsumer。
构建FlinkKafkaConsumer java示例代码如下：

Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); // only required for Kafka 0.8 properties.setProperty("zookeeper.connect", "localhost:2181"); properties.setProperty("group.id", "test"); DataStream stream = env .addSource(new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties));

scala：

val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") // only required for Kafka 0.8 properties.setProperty("zookeeper.connect", "localhost:2181") properties.setProperty("group.id", "test") stream = env .addSource(new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), properties)) .print()

必须有的：
1.topic名称
2.用于反序列化Kafka数据的DeserializationSchema / KafkaDeserializationSchema
3.配置参数：“bootstrap.servers” “group.id” （kafka0.8还需要 “zookeeper.connect”）
配置消费起始位置 java：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); FlinkKafkaConsumer myConsumer = new FlinkKafkaConsumer<>(...); myConsumer.setStartFromEarliest(); // start from the earliest record possible myConsumer.setStartFromLatest(); // start from the latest record myConsumer.setStartFromTimestamp(...); // start from specified epoch timestamp (milliseconds) myConsumer.setStartFromGroupOffsets(); // the default behaviour//指定位置 //Map specificStartOffsets = new HashMap<>(); //specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L); //myConsumer.setStartFromSpecificOffsets(specificStartOffsets); DataStream stream = env.addSource(myConsumer);

scala：

val env = StreamExecutionEnvironment.getExecutionEnvironment()val myConsumer = new FlinkKafkaConsumer[String](...) myConsumer.setStartFromEarliest()// start from the earliest record possible myConsumer.setStartFromLatest()// start from the latest record myConsumer.setStartFromTimestamp(...)// start from specified epoch timestamp (milliseconds) myConsumer.setStartFromGroupOffsets()// the default behaviour//指定位置 //val specificStartOffsets = new java.util.HashMap[KafkaTopicPartition, java.lang.Long]() //specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L) //myConsumer.setStartFromSpecificOffsets(specificStartOffsets)val stream = env.addSource(myConsumer)

检查点 启用Flink的检查点后，Flink Kafka Consumer将使用主题中的记录，并以一致的方式定期检查其所有Kafka偏移以及其他操作的状态。如果作业失败，Flink会将流式程序恢复到最新检查点的状态，并从存储在检查点中的偏移量开始重新使用Kafka的记录。
如果禁用了检查点，则Flink Kafka Consumer依赖于内部使用的Kafka客户端的自动定期偏移提交功能。
如果启用了检查点，则Flink Kafka Consumer将在检查点完成时提交存储在检查点状态中的偏移量。
java

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

scala

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

分区发现 Flink Kafka Consumer支持发现动态创建的Kafka分区，并使用一次性保证消费它们。
还可以使用正则：
java

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("group.id", "test"); FlinkKafkaConsumer011 myConsumer = new FlinkKafkaConsumer011<>( java.util.regex.Pattern.compile("test-topic-[0-9]"), new SimpleStringSchema(), properties); DataStream stream = env.addSource(myConsumer); ...

scala

val env = StreamExecutionEnvironment.getExecutionEnvironment()val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") properties.setProperty("group.id", "test")val myConsumer = new FlinkKafkaConsumer08[String]( java.util.regex.Pattern.compile("test-topic-[0-9]"), new SimpleStringSchema, properties)val stream = env.addSource(myConsumer) ...

时间戳和水印 在许多情况下，记录的时间戳（显式或隐式）嵌入记录本身。另外，用户可能想要周期性地或以不规则的方式发出水印。
我们可以定义好Timestamp Extractors / Watermark Emitters，通过以下方式将其传递给您的消费者：
java

Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); // only required for Kafka 0.8 properties.setProperty("zookeeper.connect", "localhost:2181"); properties.setProperty("group.id", "test"); FlinkKafkaConsumer08 myConsumer = new FlinkKafkaConsumer08<>("topic", new SimpleStringSchema(), properties); myConsumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter()); DataStream stream = env .addSource(myConsumer) .print();

scala

val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") // only required for Kafka 0.8 properties.setProperty("zookeeper.connect", "localhost:2181") properties.setProperty("group.id", "test")val myConsumer = new FlinkKafkaConsumer08[String]("topic", new SimpleStringSchema(), properties) myConsumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter()) stream = env .addSource(myConsumer) .print()

Kafka Producer Kafka Producer 根据版本分别叫做FlinkProducer011 FlinkKafkaProducer010等等
Kafka >= 1.0.0 的版本就叫FlinkKafkaProducer 。
构建FlinkKafkaConsumer java

DataStream stream = ...; FlinkKafkaProducer011 myProducer = new FlinkKafkaProducer011( "localhost:9092",// broker list "my-topic",// target topic new SimpleStringSchema()); // serialization schema// versions 0.10+ allow attaching the records' event timestamp when writing them to Kafka; // this method is not available for earlier Kafka versions myProducer.setWriteTimestampToKafka(true); stream.addSink(myProducer);

scala

val stream: DataStream[String] = ...val myProducer = new FlinkKafkaProducer011[String]( "localhost:9092",// broker list "my-topic",// target topic new SimpleStringSchema)// serialization schema// versions 0.10+ allow attaching the records' event timestamp when writing them to Kafka; // this method is not available for earlier Kafka versions myProducer.setWriteTimestampToKafka(true)stream.addSink(myProducer)

需要指定broker list ， topic，序列化类。
自定义分区：默认情况下，将使用FlinkFixedPartitioner将每个Flink Kafka Producer并行子任务映射到单个Kafka分区。
可以实现FlinkKafkaPartitioner类自定义分区。
Flink1.9消费Kafka完整代码：

import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import java.util.Properties; public class KafkaConsumer {public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); properties.setProperty("group.id", "test"); //构建FlinkKafkaConsumer FlinkKafkaConsumer myConsumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties); //指定偏移量 myConsumer.setStartFromEarliest(); DataStream stream = env .addSource(myConsumer); env.enableCheckpointing(5000); stream.print(); env.execute("Flink Streaming Java API Skeleton"); }

项目地址：https://github.com/tree1123/flink_demo_1.9
更多Flink知识，欢迎关注实时流式计算
文章图片

转载于:https://www.cnblogs.com/tree1123/p/11556114.html

推荐阅读

• 

  《时生》
• 

  中国为何不承认科索沃 北约轰炸南联盟
• 

  空调内外机通风口是什么
• 

  羊肉变质了煮熟还能吃吗
• 

  FGO贝奥武夫三技能强化简评 FGO圣诞六期贝奥武夫技能本
• 

  秦时明月世界测试福利有什么 秦时明月世界测试福利介绍
• 

  redis connect time out redis连接超时功能
• 

  穿衬衫要不要穿外套 穿衬衫里面还需要穿衣服吗
• 

  藜麦碳水化合物含量高吗
• 

  爱目者护眼贴关爱儿童眼部健康公益活动|爱目者护眼贴关爱儿童眼部健康公益活动 共筑美好未来
• 

  sp如何调玻璃材质 sp如何diy，sp如何diy更疼
• 

  鼻毛|鼻毛外露不美观，拔了完事？鼻毛好处很多，劝你手下留情
• 

  你们看过后觉得最好的小说是什么？有哪些“高干文”和“糙汉文”推荐？
• 

  美丽加
• 

  电商什么都卖好吗怎么说 电商什么都卖好吗，电商什么都卖好吗知乎
• 

  地不容价格 地不容
• 

  八月瓜种植技术及条件
• 

  笼养八哥怎么训练 怎样饲养和调教八哥
• 

  佳能A4000 IS价格是多少钱？佳能A4000 IS的佳能A4000 IS说明书
• 

  天天吃咸鸭蛋好吗会胖吗 天天吃咸鸭蛋好吗

• 深入浅出谈一下有关分布式消息技术（Kafka）
• Spring|Spring Boot 整合 Activiti6.0.0
• 15.Kafka
• springboot整合数据库连接池-->druid
• SpringBoot整合MongoDB完整实例代码
• Springboot整合kafka的示例代码
• 15天蜕变之旅Day6～5月29日作业父母能量链接内在整合
• Spring|Spring Boot整合Mybatis进行CRUD
• Springboot整合RabbitMQ（三）——Topic主题交换机
• 搭建大数据三节点（Hadoop、Hbase、Zookeeper、Kafka、Hive）环境（含配置文件参考）