Flink-Kafka-Connector Flink结合Kafka实战
简介
Flink-kafka-connector用来做什么?
Kafka中的partition机制和Flink的并行度机制结合,实现数据恢复
Kafka可以作为Flink的source和sink
任务失败,通过设置kafka的offset来恢复应用
kafka简单介绍
关于kafka,我们会有专题文章介绍,这里简单介绍几个必须知道的概念。
1.生产者(Producer)
顾名思义,生产者就是生产消息的组件,它的主要工作就是源源不断地生产出消息,然后发送给消息队列。生产者可以向消息队列发送各种类型的消息,如狭义的字符串消息,也可以发送二进制消息。生产者是消息队列的数据源,只有通过生产者持续不断地向消息队列发送消息,消息队列才能不断处理消息。
2.消费者(Consumer)
所谓消费者,指的是不断消费(获取)消息的组件,它获取消息的来源就是消息队列(即Kafka本身)。换句话说,生产者不断向消息队列发送消息,而消费者则不断从消息队列中获取消息。
3.主题(Topic)
主题是Kafka中一个极为重要的概念。首先,主题是一个逻辑上的概念,它用于从逻辑上来归类与存储消息本身。多个生产者可以向一个Topic发送消息,同时也可以有多个消费者消费一个Topic中的消息。Topic还有分区和副本的概念。Topic与消息这两个概念之间密切相关,Kafka中的每一条消息都归属于某一个Topic,而一个Topic下面可以有任意数量的消息。
kafka简单操作
启动zk:nohup bin/zookeeper-server-start.sh config/zookeeper.properties &
启动server: nohup bin/kafka-server-start.sh config/server.properties &
创建一个topic:bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
查看topic:bin/kafka-topics.sh --list --zookeeper localhost:2181
发送数据:bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
启动一个消费者:bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
删除topic: bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topn
Flink消费Kafka注意事项
-
setStartFromGroupOffsets()【默认消费策略】
默认读取上次保存的offset信息 如果是应用第一次启动,读取不到上次的offset信息,则会根据这个参数auto.offset.reset的值来进行消费数据
- setStartFromEarliest()
从最早的数据开始进行消费,忽略存储的offset信息
- setStartFromLatest()
从最新的数据进行消费,忽略存储的offset信息
- setStartFromSpecificOffsets(Map)
从指定位置进行消费 - 当checkpoint机制开启的时候,KafkaConsumer会定期把kafka的offset信息还有其他operator的状态信息一块保存起来。当job失败重启的时候,Flink会从最近一次的checkpoint中进行恢复数据,重新消费kafka中的数据。
- 为了能够使用支持容错的kafka Consumer,需要开启checkpoint
env.enableCheckpointing(5000); // 每5s checkpoint一次
搭建Kafka单机环境
我本地安装了一个kafka_2.11-2.1.0版本的kafka
启动Zookeeper和kafka server:
启动zk:nohup bin/zookeeper-server-start.sh config/zookeeper.properties & 启动server: nohup bin/kafka-server-start.sh config/server.properties &
创建一个topic:
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
实战案例
Kafka作为Flink Sink
首先pom依赖:
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka_2.11</artifactId> <version>1.7.0</version> </dependency>
向kafka写入数据:
public class KafkaProducer { public static void main(String[] args) throws Exception{ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStreamSource<String> text = env.addSource(new MyNoParalleSource()).setParallelism(1); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); //new FlinkKafkaProducer("topn",new KeyedSerializationSchemaWrapper(new SimpleStringSchema()),properties,FlinkKafkaProducer.Semantic.EXACTLY_ONCE); FlinkKafkaProducer<String> producer = new FlinkKafkaProducer("test",new SimpleStringSchema(),properties); /* //event-timestamp事件的发生时间 producer.setWriteTimestampToKafka(true); */ text.addSink(producer); env.execute(); } }//
大家这里特别注意,我们实现了一个并行度为1的MyNoParalleSource
来生产数据,代码如下:
//使用并行度为1的source public class MyNoParalleSource implements SourceFunction<String> {//1 //private long count = 1L; private boolean isRunning = true; /** * 主要的方法 * 启动一个source * 大部分情况下,都需要在这个run方法中实现一个循环,这样就可以循环产生数据了 * * @param ctx * @throws Exception */ @Override public void run(SourceContext<String> ctx) throws Exception { while(isRunning){ //图书的排行榜 List<String> books = new ArrayList<>(); books.add("Pyhton从入门到放弃");//10 books.add("Java从入门到放弃");//8 books.add("Php从入门到放弃");//5 books.add("C++从入门到放弃");//3 books.add("Scala从入门到放弃");//0-4 int i = new Random().nextInt(5); ctx.collect(books.get(i)); //每2秒产生一条数据 Thread.sleep(2000); } } //取消一个cancel的时候会调用的方法 @Override public void cancel() { isRunning = false; } }
代码实现了一个发送器,来发送书名等...
然后右键运行我们的程序,控制台输出如下:
开始源源不断的生产数据了。
然后我们用命令去查看一下 kafka test
这个topic:
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
输出如下:
Kafka作为Flink Source
直接上代码:
public class KafkaConsumer { public static void main(String[] args) throws Exception{ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "localhost:9092"); FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties); //从最早开始消费 consumer.setStartFromEarliest(); DataStream<String> stream = env .addSource(consumer); stream.print(); //stream.map(); env.execute(); } }//
控制台输出如下:
将我们之前发往kafka的消息全部打印出来了。
低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。
持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。
转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。
- 上一篇
Flink广播变量
广播变量简介 在Flink中,同一个算子可能存在若干个不同的并行实例,计算过程可能不在同一个Slot中进行,不同算子之间更是如此,因此不同算子的计算数据之间不能像Java数组之间一样互相访问,而广播变量Broadcast便是解决这种情况的。 我们可以把广播变量理解为是一个公共的共享变量,我们可以把一个dataset 数据集广播出去,然后不同的task在节点上都能够获取到,这个数据在每个节点上只会存在一份 用法 1:初始化数据 DataSet<Integer> num = env.fromElements(1, 2, 3) 2:广播数据 .withBroadcastSet(toBroadcast, "num"); 3:获取数据 Collection<Integer> broadcastSet = getRuntimeContext().getBroadcastVariable("num"); 注意事项 使用广播状态,task 之间不会相互通信 只有广播的一边可以修改广播状态的内容。用户必须保证所有 operator 并发实例上对广播状态的 修改行为都是一致的。或者...
- 下一篇
Flink-Table-&-SQL
简介 Apache Flink具有两个关系API - 表API和SQL - 用于统一流和批处理。Table API是Scala和Java的语言集成查询API,允许以非常直观的方式组合来自关系运算符的查询,Table API和SQL接口彼此紧密集成,以及Flink的DataStream和DataSet API。您可以轻松地在基于API构建的所有API和库之间切换。例如,您可以使用CEP库从DataStream中提取模式,然后使用Table API分析模式,或者可以在预处理上运行Gelly图算法之前使用SQL查询扫描,过滤和聚合批处理表数据。 Flink SQL的编程模型 创建一个TableEnvironment TableEnvironment是Table API和SQL集成的核心概念,它主要负责:1、在内部目录中注册一个Table2、注册一个外部目录3、执行SQL查询4、注册一个用户自定义函数(标量、表及聚合)5、将DataStream或者DataSet转换成Table6、持有ExecutionEnvironment或者StreamExecutionEnvironment的引用一个Ta...
相关文章
文章评论
共有0条评论来说两句吧...
文章二维码
点击排行
推荐阅读
最新文章
- SpringBoot2整合Redis,开启缓存,提高访问速度
- CentOS关闭SELinux安全模块
- CentOS7编译安装Cmake3.16.3,解决mysql等软件编译问题
- Docker快速安装Oracle11G,搭建oracle11g学习环境
- Jdk安装(Linux,MacOS,Windows),包含三大操作系统的最全安装
- 设置Eclipse缩进为4个空格,增强代码规范
- MySQL8.0.19开启GTID主从同步CentOS8
- SpringBoot2整合MyBatis,连接MySql数据库做增删改查操作
- Docker安装Oracle12C,快速搭建Oracle学习环境
- CentOS7安装Docker,走上虚拟化容器引擎之路