NiFi 入门:Kakfa 数据同步到关系数据库(PostgreSQL)--Part3

栏目: 数据库 · PostgreSQL · 发布时间: 6年前

Part1 和Part2把如何使用EvaluateJsonPath获取属性值,SplitJson 拆分行数据说清楚了,还差Kafka源的部分.

Kafka源的设置:

NiFi中对应Kafka有不同版本, 分队对应0.9,0.10,0.11和1.0的不同Kafka版本,低版本的process可以处理高版本的Kafka

NiFi 入门:Kakfa 数据同步到关系数据库(PostgreSQL)--Part3

HDP用的Kafka0.10的版本,我使用了ConsumeKafka_0_10,注意groupid不用和其他用的process重复,不然会被消费掉.

NiFi 入门:Kakfa 数据同步到关系数据库(PostgreSQL)--Part3

进一步思考:

1. 判断记录是否符合格式

增加了属性判断:使用RouteOnAttribute判断trx_num属性是否非空

${trx_num:isEmpty():not()}

防止完成后的概览图:

NiFi 入门:Kakfa 数据同步到关系数据库(PostgreSQL)--Part3

2.考虑使用record模式

attribute模式给了用户很多自由度,但是也导致了,属性要多次配置,record模式比较好的解决了这个模式,通过统一定义schema,

从Kafka端就可以很好的解构

3. flowfile的合并

现在putsql是一行一行的作insert, 记录数不多还好,多的话对性能影响较大.并且如果是往其他目标insert(如hdfs\mogndo)等, 需要配合mergercontent组件完成.

本文由seng 创作,采用 知识共享署名-相同方式共享 3.0 中国大陆许可协议 进行许可。

转载、引用前需联系作者,并署名作者且注明文章出处。

本站文章版权归原作者及原出处所有 。内容为作者个人观点, 并不代表本站赞同其观点和对其真实性负责。本站是一个个人学习交流的平台,并不用于任何商业目的,如果有任何问题,请及时联系我们,我们将根据著作权人的要求,立即更正或者删除有关内容。本站拥有对此声明的最终解释权。


以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们

Hit Refresh

Hit Refresh

Satya Nadella、Greg Shaw / HarperBusiness / 2017-9-26 / USD 20.37

Hit Refresh is about individual change, about the transformation happening inside of Microsoft and the technology that will soon impact all of our lives—the arrival of the most exciting and disruptive......一起来看看 《Hit Refresh》 这本书的介绍吧!

HTML 压缩/解压工具
HTML 压缩/解压工具

在线压缩/解压 HTML 代码

Base64 编码/解码
Base64 编码/解码

Base64 编码/解码

SHA 加密
SHA 加密

SHA 加密工具