Alink连接Kafka数据源(Java版本)
本文主要讨论如何使用Alink的Kafka连接组件(Kafka011SourceStreamOp和Kafka011SinkStreamOp)读取写入数据。如何你需要一个本地的Kafka数据源进行实验,可以参考我另外一篇文章,详细介绍了搭建Kafka及建立Topic的过程。
Alink品数:在MacOS上搭建Kafkazhuanlan.zhihu.comAlink品数:在Windo上搭建Kafkazhuanlan.zhihu.com
首先,我们演示如何将流式数据写入Kafka。
假设已经有一个Kafka的数据源(譬如:本地Kafka数据源,端口为9092),并且Kafka中已经有一个topic,名称为iris,则Kafka写入组件Kafka011SinkStreamOp可以如下设置:
Kafka011SinkStreamOp sink = new Kafka011SinkStreamOp().setBootstrapServers("localhost:9092").setDataFormat("json").setTopic("iris");
注意:Kafka写入的数据只能为字符串,需要设置每条记录转化为字符串的方式,这里我们使用Json格式。
我们还需要构造一个获取流式数据的方式,最简单的方式是使用CsvSourceStreamOp组件,将csv数据(https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv)以流的方式读入。然后,再连接Kafka写入组件,开始执行流式操作。完整代码如下:
private static void writeKafka() throws Exception {String URL = "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv";String SCHEMA_STR= "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";CsvSourceStreamOp data = new CsvSourceStreamOp().setFilePath(URL).setSchemaStr(SCHEMA_STR);Kafka011SinkStreamOp sink = new Kafka011SinkStreamOp().setBootstrapServers("localhost:9092").setDataFormat("json").setTopic("iris");data.link(sink);StreamOperator.execute();}
由于CSV文件中数据有限,当读取完最后一条时,流式任务会结束。
接下来,我们可以使用Alink的Kafka011SourceStreamOp组件读取数据,并设置其消费者组ID,读取模式为从头开始,具体代码如下:
private static void readKafka() throws Exception {Kafka011SourceStreamOp source = new Kafka011SourceStreamOp().setBootstrapServers("localhost:9092").setTopic("iris").setStartupMode("EARLIEST").setGroupId("alink_group");source.print();StreamOperator.execute();}
执行打印结果如下,中间略去大部分数据:
message_key|message|topic|topic_partition|partition_offset
-----------|-------|-----|---------------|----------------
null|{"sepal_width":3.4,"petal_width":0.2,"sepal_length":4.8,"category":"Iris-setosa","petal_length":1.6}|iris|0|0
null|{"sepal_width":4.1,"petal_width":0.1,"sepal_length":5.2,"category":"Iris-setosa","petal_length":1.5}|iris|0|1
null|{"sepal_width":2.8,"petal_width":1.5,"sepal_length":6.5,"category":"Iris-versicolor","petal_length":4.6}|iris|0|2
null|{"sepal_width":3.0,"petal_width":1.8,"sepal_length":6.1,"category":"Iris-virginica","petal_length":4.9}|iris|0|3
null|{"sepal_width":2.9,"petal_width":1.8,"sepal_length":7.3,"category":"Iris-virginica","petal_length":6.3}|iris|0|4
......
null|{"sepal_width":2.2,"petal_width":1.0,"sepal_length":6.0,"category":"Iris-versicolor","petal_length":4.0}|iris|0|145
null|{"sepal_width":2.4,"petal_width":1.0,"sepal_length":5.5,"category":"Iris-versicolor","petal_length":3.7}|iris|0|146
null|{"sepal_width":3.1,"petal_width":0.2,"sepal_length":4.6,"category":"Iris-setosa","petal_length":1.5}|iris|0|147
null|{"sepal_width":3.4,"petal_width":0.2,"sepal_length":4.8,"category":"Iris-setosa","petal_length":1.9}|iris|0|148
null|{"sepal_width":2.9,"petal_width":1.4,"sepal_length":6.1,"category":"Iris-versicolor","petal_length":4.7}|iris|0|149
可以看到直接从Kafka中获取的每条数据都是Json格式的字符串。
接下来,我们需要对字符串里面的数据进行提取。推荐使用JsonValueStreamOp,通过设置需要提取内容的JsonPath,提取出各列数据。详细代码如下:
Kafka011SourceStreamOp source =new Kafka011SourceStreamOp().setBootstrapServers("localhost:9092").setTopic("iris").setStartupMode("EARLIEST").setGroupId("alink_group");StreamOperator data = source.link(new JsonValueStreamOp().setSelectedCol("message").setReservedCols(new String[] {}).setOutputCols(new String[] {"sepal_length", "sepal_width", "petal_length", "petal_width", "category"}).setJsonPath(new String[] {"$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width","$.category"}));System.out.print(data.getSchema());data.print();StreamOperator.execute();
关于结果数据的Schema打印为:
root|-- sepal_length: STRING|-- sepal_width: STRING|-- petal_length: STRING|-- petal_width: STRING|-- category: STRING
可以看出JsonValueStreamOp提取出来的结果都是string类型的,具体数据打印结果如下,略去中间的大部分数据。
sepal_length|sepal_width|petal_length|petal_width|category
------------|-----------|------------|-----------|--------
4.8|3.4|1.6|0.2|Iris-setosa
5.2|4.1|1.5|0.1|Iris-setosa
6.5|2.8|4.6|1.5|Iris-versicolor
6.1|3.0|4.9|1.8|Iris-virginica
7.3|2.9|6.3|1.8|Iris-virginica
......
5.2|2.7|3.9|1.4|Iris-versicolor
6.4|2.7|5.3|1.9|Iris-virginica
6.8|3.0|5.5|2.1|Iris-virginica
5.7|2.5|5.0|2.0|Iris-virginica
6.1|2.8|4.0|1.3|Iris-versicolor
至此,我们已经能够拿到数据了,只是数据的类型有问题,需要进行转换。我们可以使用Flink SQL 的cast方法,在代码实现上,只需在连接JsonValueStreamOp之后,使用select方法(其参数为SQL语句),具体代码如下:
StreamOperator data = source.link(new JsonValueStreamOp().setSelectedCol("message").setReservedCols(new String[] {}).setOutputCols(new String[] {"sepal_length", "sepal_width", "petal_length", "petal_width", "category"}).setJsonPath(new String[] {"$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width","$.category"})).select("CAST(sepal_length AS DOUBLE) AS sepal_length, "+ "CAST(sepal_width AS DOUBLE) AS sepal_width, "+ "CAST(petal_length AS DOUBLE) AS petal_length, "+ "CAST(petal_width AS DOUBLE) AS petal_width, category");
执行新的代码,关于结果数据的Schema打印为:
root|-- sepal_length: DOUBLE|-- sepal_width: DOUBLE|-- petal_length: DOUBLE|-- petal_width: DOUBLE|-- category: STRING
每列数据都转化为相应的类型。具体数据打印结果如下,略去中间的大部分数据。
sepal_length|sepal_width|petal_length|petal_width|category
------------|-----------|------------|-----------|--------
4.8000|3.4000|1.6000|0.2000|Iris-setosa
5.2000|4.1000|1.5000|0.1000|Iris-setosa
6.5000|2.8000|4.6000|1.5000|Iris-versicolor
6.1000|3.0000|4.9000|1.8000|Iris-virginica
7.3000|2.9000|6.3000|1.8000|Iris-virginica
......
5.2000|2.7000|3.9000|1.4000|Iris-versicolor
6.4000|2.7000|5.3000|1.9000|Iris-virginica
6.8000|3.0000|5.5000|2.1000|Iris-virginica
5.7000|2.5000|5.0000|2.0000|Iris-virginica
6.1000|2.8000|4.0000|1.3000|Iris-versicolor
可以看出,配合使用Alink的相关组件,可以完整地从Kafka上读取、写入数据。后面,可通过Alink的各算法组件进行深入计算。
本文主要讨论如何使用Alink的Kafka连接组件(Kafka011SourceStreamOp和Kafka011SinkStreamOp)读取写入数据。如何你需要一个本地的Kafka数据源进行实验,可以参考我另外一篇文章,详细介绍了搭建Kafka及建立Topic的过程。
Alink品数:在MacOS上搭建Kafkazhuanlan.zhihu.comAlink品数:在Windo上搭建Kafkazhuanlan.zhihu.com
首先,我们演示如何将流式数据写入Kafka。
假设已经有一个Kafka的数据源(譬如:本地Kafka数据源,端口为9092),并且Kafka中已经有一个topic,名称为iris,则Kafka写入组件Kafka011SinkStreamOp可以如下设置:
Kafka011SinkStreamOp sink = new Kafka011SinkStreamOp().setBootstrapServers("localhost:9092").setDataFormat("json").setTopic("iris");
注意:Kafka写入的数据只能为字符串,需要设置每条记录转化为字符串的方式,这里我们使用Json格式。
我们还需要构造一个获取流式数据的方式,最简单的方式是使用CsvSourceStreamOp组件,将csv数据(https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv)以流的方式读入。然后,再连接Kafka写入组件,开始执行流式操作。完整代码如下:
private static void writeKafka() throws Exception {String URL = "https://alink-release.oss-cn-beijing.aliyuncs.com/data-files/iris.csv";String SCHEMA_STR= "sepal_length double, sepal_width double, petal_length double, petal_width double, category string";CsvSourceStreamOp data = new CsvSourceStreamOp().setFilePath(URL).setSchemaStr(SCHEMA_STR);Kafka011SinkStreamOp sink = new Kafka011SinkStreamOp().setBootstrapServers("localhost:9092").setDataFormat("json").setTopic("iris");data.link(sink);StreamOperator.execute();}
由于CSV文件中数据有限,当读取完最后一条时,流式任务会结束。
接下来,我们可以使用Alink的Kafka011SourceStreamOp组件读取数据,并设置其消费者组ID,读取模式为从头开始,具体代码如下:
private static void readKafka() throws Exception {Kafka011SourceStreamOp source = new Kafka011SourceStreamOp().setBootstrapServers("localhost:9092").setTopic("iris").setStartupMode("EARLIEST").setGroupId("alink_group");source.print();StreamOperator.execute();}
执行打印结果如下,中间略去大部分数据:
message_key|message|topic|topic_partition|partition_offset
-----------|-------|-----|---------------|----------------
null|{"sepal_width":3.4,"petal_width":0.2,"sepal_length":4.8,"category":"Iris-setosa","petal_length":1.6}|iris|0|0
null|{"sepal_width":4.1,"petal_width":0.1,"sepal_length":5.2,"category":"Iris-setosa","petal_length":1.5}|iris|0|1
null|{"sepal_width":2.8,"petal_width":1.5,"sepal_length":6.5,"category":"Iris-versicolor","petal_length":4.6}|iris|0|2
null|{"sepal_width":3.0,"petal_width":1.8,"sepal_length":6.1,"category":"Iris-virginica","petal_length":4.9}|iris|0|3
null|{"sepal_width":2.9,"petal_width":1.8,"sepal_length":7.3,"category":"Iris-virginica","petal_length":6.3}|iris|0|4
......
null|{"sepal_width":2.2,"petal_width":1.0,"sepal_length":6.0,"category":"Iris-versicolor","petal_length":4.0}|iris|0|145
null|{"sepal_width":2.4,"petal_width":1.0,"sepal_length":5.5,"category":"Iris-versicolor","petal_length":3.7}|iris|0|146
null|{"sepal_width":3.1,"petal_width":0.2,"sepal_length":4.6,"category":"Iris-setosa","petal_length":1.5}|iris|0|147
null|{"sepal_width":3.4,"petal_width":0.2,"sepal_length":4.8,"category":"Iris-setosa","petal_length":1.9}|iris|0|148
null|{"sepal_width":2.9,"petal_width":1.4,"sepal_length":6.1,"category":"Iris-versicolor","petal_length":4.7}|iris|0|149
可以看到直接从Kafka中获取的每条数据都是Json格式的字符串。
接下来,我们需要对字符串里面的数据进行提取。推荐使用JsonValueStreamOp,通过设置需要提取内容的JsonPath,提取出各列数据。详细代码如下:
Kafka011SourceStreamOp source =new Kafka011SourceStreamOp().setBootstrapServers("localhost:9092").setTopic("iris").setStartupMode("EARLIEST").setGroupId("alink_group");StreamOperator data = source.link(new JsonValueStreamOp().setSelectedCol("message").setReservedCols(new String[] {}).setOutputCols(new String[] {"sepal_length", "sepal_width", "petal_length", "petal_width", "category"}).setJsonPath(new String[] {"$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width","$.category"}));System.out.print(data.getSchema());data.print();StreamOperator.execute();
关于结果数据的Schema打印为:
root|-- sepal_length: STRING|-- sepal_width: STRING|-- petal_length: STRING|-- petal_width: STRING|-- category: STRING
可以看出JsonValueStreamOp提取出来的结果都是string类型的,具体数据打印结果如下,略去中间的大部分数据。
sepal_length|sepal_width|petal_length|petal_width|category
------------|-----------|------------|-----------|--------
4.8|3.4|1.6|0.2|Iris-setosa
5.2|4.1|1.5|0.1|Iris-setosa
6.5|2.8|4.6|1.5|Iris-versicolor
6.1|3.0|4.9|1.8|Iris-virginica
7.3|2.9|6.3|1.8|Iris-virginica
......
5.2|2.7|3.9|1.4|Iris-versicolor
6.4|2.7|5.3|1.9|Iris-virginica
6.8|3.0|5.5|2.1|Iris-virginica
5.7|2.5|5.0|2.0|Iris-virginica
6.1|2.8|4.0|1.3|Iris-versicolor
至此,我们已经能够拿到数据了,只是数据的类型有问题,需要进行转换。我们可以使用Flink SQL 的cast方法,在代码实现上,只需在连接JsonValueStreamOp之后,使用select方法(其参数为SQL语句),具体代码如下:
StreamOperator data = source.link(new JsonValueStreamOp().setSelectedCol("message").setReservedCols(new String[] {}).setOutputCols(new String[] {"sepal_length", "sepal_width", "petal_length", "petal_width", "category"}).setJsonPath(new String[] {"$.sepal_length", "$.sepal_width", "$.petal_length", "$.petal_width","$.category"})).select("CAST(sepal_length AS DOUBLE) AS sepal_length, "+ "CAST(sepal_width AS DOUBLE) AS sepal_width, "+ "CAST(petal_length AS DOUBLE) AS petal_length, "+ "CAST(petal_width AS DOUBLE) AS petal_width, category");
执行新的代码,关于结果数据的Schema打印为:
root|-- sepal_length: DOUBLE|-- sepal_width: DOUBLE|-- petal_length: DOUBLE|-- petal_width: DOUBLE|-- category: STRING
每列数据都转化为相应的类型。具体数据打印结果如下,略去中间的大部分数据。
sepal_length|sepal_width|petal_length|petal_width|category
------------|-----------|------------|-----------|--------
4.8000|3.4000|1.6000|0.2000|Iris-setosa
5.2000|4.1000|1.5000|0.1000|Iris-setosa
6.5000|2.8000|4.6000|1.5000|Iris-versicolor
6.1000|3.0000|4.9000|1.8000|Iris-virginica
7.3000|2.9000|6.3000|1.8000|Iris-virginica
......
5.2000|2.7000|3.9000|1.4000|Iris-versicolor
6.4000|2.7000|5.3000|1.9000|Iris-virginica
6.8000|3.0000|5.5000|2.1000|Iris-virginica
5.7000|2.5000|5.0000|2.0000|Iris-virginica
6.1000|2.8000|4.0000|1.3000|Iris-versicolor
可以看出,配合使用Alink的相关组件,可以完整地从Kafka上读取、写入数据。后面,可通过Alink的各算法组件进行深入计算。
Alink连接Kafka数据源(Java版本)相关推荐
- Alink连接Kafka数据源(Python版本)
本文主要讨论如何使用Alink的Kafka连接组件(Kafka011SourceStreamOp和Kafka011SinkStreamOp)读取写入数据.如何你需要一个本地的Kafka数据源进行实验, ...
- [数据库druid连接池实现]--Java版本
前言 之前有介绍一篇关于自定义的数据库链接池实现,但是质量并不高.而且遭到博友的一脸嫌弃!确实,拿出来的东西就应该保证是正确的,或者质量高,否则很容易就误导了他人,显得TMD也不道德,所以今天专门重写 ...
- 最新Kafka教程(包含kafka部署与基本操作、java连接kafka、spring连接kafka以及使用springboot)
最新Kafka教程(包含kafka部署与基本操作.java连接kafka.spring连接kafka以及使用springboot) 欢迎转载,转载请注明网址:https://blog.csdn.net ...
- kafka文档(3)----0.8.2-kafka API(java版本)
原文地址: http://kafka.apache.org/documentation.html#api Apache Kafka包含新的java客户端,这些新的的客户端将取代现存的Scala客户端, ...
- Java连接MySQL8.0以上版本数据库方式
Java连接MySQL8.0以上版本数据库方式 MySQL 8.0 开始数据库相比常用的 5.X 版本发生了比较大的变化,我们在连接数据库的过程中许多地方也要发生一些变化. 总结一下,想要利用 mys ...
- Java连接MySQL8.0以上版本
Java连接MySQL8.0以上版本 MySQL和JDBC版本 连接步骤与注意事项 整个流程的代码 MySQL和JDBC版本 MySQL8.0 mysql-connector-java-8.0.13. ...
- 详细讲解如何使用Java连接Kafka构建生产者和消费者(带测试样例)
1 缘起 学习消息队列的过程中,先补习了RabbitMQ相关知识, 接着又重温了Kafka相关的知识, 发现,我并没有积累Java原生操作Kafka的文章, 只使用SpringBoot集成过Kafka ...
- kafka适配Java jdk版本
kafka学习,版本适配: kafka的版本要与jdk的版本做适配,精确到小版本号才可以 我亲测的版本适配如下 kafka jdk kafka_2.13-2.5.0 jdk-8u271-linux-x ...
- kafka partition java,kafka中partition数量与消费者对应关系以及Java实践
kafka中partition数量与消费者对应关系以及Java实践 kafka中partition数量与消费者对应关系以及Java实践 kafka是由Apache软件基金会开发的一个开源流处理平台.k ...
最新文章
- 求环形数组的最大子数组的和
- android中Invalidate和postInvalidate的区别
- JavaScript onerror事件
- mysql 分时统计_mysql中数据统计的技巧备忘录
- Objective-C 文件夹操作
- python中乘法和除法_python – NumPy的性能:uint8对比浮动和乘法与除法?
- Nginx----原理
- TCP/IP协议、DoD模型、OSI模型
- rabbitmq页面出现/etc/rabbitmq/rabbitmq.config(not found)解决方法
- python多维数据聚类可视化_基于python3的可视化数据聚类系统(k-means算法和k-中心点算法)...
- h5居中loading_H5样式与布局 --常用居中方法
- PSV 基本的游戏下载方式及局部功能介绍
- 微信用户扫码登录和登录退出的业务逻辑实现(java版)
- 字典(数字大小写转换器)
- Oracle in 不能超过1000的解决方案
- 为知笔记如何一键收藏微信文章?
- uni-app自定义全屏切换组件
- 【校内模拟】简单粗暴的题目(二项式展开)
- k短路(HDU6705)
- 海尔微型计算机硬盘如何拆卸,海尔xqb507288拆解图
热门文章
- Zernike多项式的Matlab代码
- Fake_AP模式下的Easy-Creds浅析
- windows7C盘减少,瘦身
- 千里之堤,毁于蚁穴----考试系统
- python中invalid argument_Python创建文件报错OSError:[Errno 22] Invalid argument处理
- [Java]_[初级]_[使用正则高效替换字符串的多个占位符为多个值]
- Glide加载图片框架(仅获取BitMap)
- 无线“蹭网卡”热卖 任意密码5分钟破解
- 干货 | 美团如何基于深度学习实现图像的智能审核?
- IaaS首席架构师的架构设计思考与实践