kafka maven 依赖_Flink的sink实战之二:kafka
本文是《Flink的sink实战》系列的第二篇,《Flink的sink实战之一:初探》对sink有了基本的了解,本章来体验将数据sink到kafka的操作;
版本和环境准备
本次实战的环境和版本如下:
- JDK:1.8.0_211
- Flink:1.9.2
- Maven:3.6.0
- 操作系统:macOS Catalina 10.15.3 (MacBook Pro 13-inch, 2018)
- IDEA:2018.3.5 (Ultimate Edition)
- Kafka:2.4.0
- Zookeeper:3.5.5
请确保上述环境和服务已经就绪;
源码下载
如果您不想写代码,整个系列的源码可在GitHub下载到,地址和链接信息如下表所示:
这个git项目中有多个文件夹,本章的应用在flinksinkdemo文件夹下,如下图红框所示:
准备工作
正式编码前,先去官网查看相关资料了解基本情况:
地址:https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/connectors/kafka.html我这里用的kafka是2.4.0版本,在官方文档查找对应的库和类,如下图红框所示:
kafka准备
- 创建名为test006的topic,有四个分区,参考命令:
./kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --replication-factor 1 --partitions 4 --topic test006
- 在控制台消费test006的消息,参考命令:
./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test006
- 此时如果该topic有消息进来,就会在控制台输出;
- 接下来开始编码;
创建工程
- 用maven命令创建flink工程:
mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.9.2
- 根据提示,groupid输入com.bolingcavalry,artifactid输入flinksinkdemo,即可创建一个maven工程;
- 在pom.xml中增加kafka依赖库:
org.apache.flink flink-connector-kafka_2.11 1.9.0
- 工程创建完成,开始编写flink任务的代码;
发送字符串消息的sink
先尝试发送字符串类型的消息:
- 创建KafkaSerializationSchema接口的实现类,后面这个类要作为创建sink对象的参数使用:
package com.bolingcavalry.addsink;import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;import org.apache.kafka.clients.producer.ProducerRecord;import java.nio.charset.StandardCharsets;public class ProducerStringSerializationSchema implements KafkaSerializationSchema { private String topic; public ProducerStringSerializationSchema(String topic) { super(); this.topic = topic; } @Override public ProducerRecord serialize(String element, Long timestamp) { return new ProducerRecord(topic, element.getBytes(StandardCharsets.UTF_8)); }}
- 创建任务类KafkaStrSink,请注意FlinkKafkaProducer对象的参数,FlinkKafkaProducer.Semantic.EXACTLY_ONCE表示严格一次:
package com.bolingcavalry.addsink;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;import java.util.ArrayList;import java.util.List;import java.util.Properties;public class KafkaStrSink { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //并行度为1 env.setParallelism(1); Properties properties = new Properties(); properties.setProperty("bootstrap.servers", "192.168.50.43:9092"); String topic = "test006"; FlinkKafkaProducer producer = new FlinkKafkaProducer<>(topic, new ProducerStringSerializationSchema(topic), properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE); //创建一个List,里面有两个Tuple2元素 List list = new ArrayList<>(); list.add("aaa"); list.add("bbb"); list.add("ccc"); list.add("ddd"); list.add("eee"); list.add("fff"); list.add("aaa"); //统计每个单词的数量 env.fromCollection(list) .addSink(producer) .setParallelism(4); env.execute("sink demo : kafka str"); }}
使用mvn命令编译构建,在target目录得到文件flinksinkdemo-1.0-SNAPSHOT.jar;在flink的web页面提交flinksinkdemo-1.0-SNAPSHOT.jar,并制定执行类,如下图:
- 提交成功后,如果flink有四个可用slot,任务会立即执行,会在消费kafak消息的终端收到消息,如下图:
- 任务执行情况如下图:
发送对象消息的sink
再来尝试如何发送对象类型的消息,这里的对象选择常用的Tuple2对象:
- 创建KafkaSerializationSchema接口的实现类,该类后面要用作sink对象的入参,请注意代码中捕获异常的那段注释:生产环境慎用printStackTrace()!!!
package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;import org.apache.kafka.clients.producer.ProducerRecord;import javax.annotation.Nullable;public class ObjSerializationSchema implements KafkaSerializationSchema> { private String topic; private ObjectMapper mapper; public ObjSerializationSchema(String topic) { super(); this.topic = topic; } @Override public ProducerRecord serialize(Tuple2 stringIntegerTuple2, @Nullable Long timestamp) { byte[] b = null; if (mapper == null) { mapper = new ObjectMapper(); } try { b= mapper.writeValueAsBytes(stringIntegerTuple2); } catch (JsonProcessingException e) { // 注意,在生产环境这是个非常危险的操作, // 过多的错误打印会严重影响系统性能,请根据生产环境情况做调整 e.printStackTrace(); } return new ProducerRecord(topic, b); }}
- 创建flink任务类:
package com.bolingcavalry.addsink;import org.apache.flink.api.java.tuple.Tuple2;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;import java.util.ArrayList;import java.util.List;import java.util.Properties;public class KafkaObjSink { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //并行度为1 env.setParallelism(1); Properties properties = new Properties(); //kafka的broker地址 properties.setProperty("bootstrap.servers", "192.168.50.43:9092"); String topic = "test006"; FlinkKafkaProducer> producer = new FlinkKafkaProducer<>(topic, new ObjSerializationSchema(topic), properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE); //创建一个List,里面有两个Tuple2元素 List> list = new ArrayList<>(); list.add(new Tuple2("aaa", 1)); list.add(new Tuple2("bbb", 1)); list.add(new Tuple2("ccc", 1)); list.add(new Tuple2("ddd", 1)); list.add(new Tuple2("eee", 1)); list.add(new Tuple2("fff", 1)); list.add(new Tuple2("aaa", 1)); //统计每个单词的数量 env.fromCollection(list) .keyBy(0) .sum(1) .addSink(producer) .setParallelism(4); env.execute("sink demo : kafka obj"); }}
- 像前一个任务那样编译构建,把jar提交到flink,并指定执行类是com.bolingcavalry.addsink.KafkaObjSink;
- 消费kafka消息的控制台输出如下:
- 在web页面可见执行情况如下:
至此,flink将计算结果作为kafka消息发送出去的实战就完成了,希望能给您提供参考,接下来的章节,我们会继续体验官方提供的sink能力;
欢迎关注公众号:程序员欣宸
kafka maven 依赖_Flink的sink实战之二:kafka相关推荐
- kafka 的pom文件_Flink的sink实战之二:kafka
欢迎访问我的GitHub https://github.com/zq2599/blog_demos 内容:所有原创文章分类汇总及配套源码,涉及Java.Docker.Kubernetes.DevOPS ...
- kafka maven 依赖_SpringBoot入门建站全系列(二十八)整合Kafka做日志监控
SpringBoot入门建站全系列(二十八)整合Kafka做日志监控 一.概述 Apache Kafka是一个分布式发布 - 订阅消息系统和一个强大的队列,可以处理大量的数据,并使您能够将消息从一个端 ...
- kafka maven没有下载_Kafka 系列(三)——Kafka 生产者详解
首先介绍一下 Kafka 生产者发送消息的过程: Kafka 会将发送消息包装为 ProducerRecord 对象, ProducerRecord 对象包含了目标主题和要发送的内容,同时还可以指定键 ...
- Flink的sink实战之一:初探,2020-2021蚂蚁金服Java面试真题解析
关于<Flink的sink实战>系列文章 本文是<Flink的sink实战>的第一篇,旨在初步了解sink,通过对基本API和addSink方法的分析研究,为后续的编码实战打好 ...
- 【华为云实战开发】15.Maven依赖的JAR包下载慢?赶紧看过来
文:华为云DevCloud 亲亲土豆 相信许多JAVA开发者在日常工作中时常会碰到这种情况,那就是编译Maven工程时,工程所依赖的jar包文件下载非常慢,甚至经常出现下载不成功的问题,今天,小编就给 ...
- 一次Maven依赖冲突采坑,把依赖调解、类加载彻底整明白了
今年年初的时候,阅读过<Maven实战>,当时有了解到Maven可以依赖调解,即当包版本不一致时,会根据一定规则选择相应的包来加载,从而避免冲突.当时不解的是既然Maven都能解决冲突,为 ...
- Maven 依赖冲突踩坑后,将依赖调解、类加载彻底整明白了
作者 | 草捏子 来源 | 草捏子(ID:chaycao) 头图 | CSDN 下载自东方IC 今年年初的时候,阅读过<Maven实战>,当时有了解到Maven可以依赖调解,即当包版本不 ...
- Maven依赖详细理解
Maven依赖理解 1 简介 2 依赖的配置 3 依赖的范围 4 传递性依赖 4.1 传递性依赖和依赖调解 5 依赖调解 6 可选依赖 7 总结 8 下载 1 简介 在Maven项目中,有一个核心文 ...
- kafka(组件分析 整合springboot集成 实战)
kafka 组件 搭建 springboot集成 实战 kafka 组件 搭建 springboot集成 实战 1.应用场景 1.1 kafka场景 1.2 kafka特性 1.3 消息对比 1.4 ...
最新文章
- boost::mp11::mp_map_update_q相关用法的测试程序
- php实现的购物车类实例,PHP实现的购物车类实例_PHP
- c语言中热河输入空格,承德市2020年(春秋版)小学英语六年级上册期中考试模拟试卷(1)C卷...
- ArcGIS License启动无响应
- Photoshop一些人像处理技巧总结
- Mybatis框架的连接池配置
- Base64编码及其作用
- python自动化部署工具_Python + Allure(报告)+ Jenkins(持续集成)接口自动化测试环境搭建...
- OpenGL ES总结(五)OpenGL ES中pipeLine机制
- 2021-06-28操作表单
- JetBrains DataGrip 2018.2.3中文破解版 含jar文件注册码激活教程(转)
- Linux之查看物理主机的CPU温度
- RESTful Web 服务:教程
- 【安全知识分享】工程项目夏季、雨季、夜间、防汛施工安全培训教育(附下载)
- win10 1809无法更新(扩展EFI分区)
- 双通道IPC如何RTSP取两个通道视频流?
- 电视常用接口(TV,AV,S-Video,YCbCr/PCbCr,VGA,Scart,DVI/HDMI)
- 一体化伺服电机编码器值清零或设置原点如何操作?
- 甜椒刷机助手(安卓一键刷机助手) v3.5.1.1 电脑版
- testtesttesttesttesttesttest
热门文章
- electron 前端开发桌面应用
- 【BZOJ3884】上帝与集合的正确用法
- 取出字符串中的数字部分
- iOS 设计中 网络请求之 同步请求(json 请求--新方法和老方法)
- [转载]qt信号signal和槽slot机制
- 备忘:SharePoint默认的欢迎WebPart中超链接样式
- 最大素数的c语言,for语句计算输出10000以内最大素数怎么搞最简单??各位大神们...
- 二分法以及random函数使用
- 水面反光如何拍摄_拍照反光太难看?这八个办法值得一试!
- Python概率编程库PyMC应用案例二则