带你体验Apache NIFI新建数据同步流程(NIFI入门)

@阿诚

发布于 2020-09-01 07:38:10

4K00

代码可运行

文章被收录于专栏：Panda诚Panda诚

运行总次数：0

代码可运行

初衷：对于一些新接触Apache NIFI的小伙伴来说，他们急于想体验NIFI，恨不得直接找到一篇文章，照着做就直接能够解决目前遇到的需求或者问题，回想当初的我，也是这个心态。其实这样的心态是不对的。好多加入NIFI学习群的新手同学都会有这个问题，一些基本的概念和知识点都没有掌握，然后提出了一堆很初级的问题，对于这些问题，我们可能已经回答了几十上百次，厌倦了，所以大家一般会说"你先去看文档吧！"。其实，对于一个新手，直接看文档，也是一脸懵。所以在这里，我带领新手的你，新建一个同步的流程，并尽可能在新建流程的同时，穿插一些基本概念。跟随本文一起操作或者只是看看，最后你可能就找到了入门的感觉了。

环境

Apache NIFI：1.11.4 单节点
Linux：CentOS Linux release 7.5.1804 16G内存 48G存储虚拟机
DB：MySQL

Apache NIFI 部署

获取安装包

我是直接编译的1.11.4的源码获取的安装包，将zip包上传到Linux，解压即可。如何下载源码及编译看编译NIFI源码
通过官方网站下载对应的NIFI安装包:https://nifi.apache.org/download.html
通过国内镜像来下载最新发布的NIFI安装包: 北理镜像(http://mirror.bit.edu.cn/apache/nifi/) 清华镜像(https://mirrors.tuna.tsinghua.edu.cn/apache/nifi/)

如果不是编译源码，建议使用国内镜像去下载，一般直接下载zip包就可以了，加入最新版本时1.11.4，那就下载nifi-1.11.4-bin.zip

部署Apache NIFI

上传Apache NIFI包到Linux上，解压安装包；或者将你的本地作为服务器，直接解压zip包。
在解压的目录下，找到conf目录，编辑bootstrap.conf文件，修改NIFI的内存配置，默认的值比较小，比如这里我改成启动2g，最大10g

java.arg.2=-Xms2g
java.arg.3=-Xmx10g

在解压的目录下，找到bin目录，可以看到里面有一些脚本

dump-nifi.bat  
nifi-env.bat  
nifi-env.sh  
nifi.sh         
run-nifi.bat  
status-nifi.bat

Linux或者Mac，使用nifi.sh start启动NIFI，nifi.sh stop停止NIFI，nifi.sh restart重启NIFI。

Windows下，直接双击run-nifi.bat即可，退出的时候关闭运行窗口就可以了。

在解压的目录下，找到log目录，当看到类似于下面的内容时，NIFI就启动成功了

2020-05-27 14:26:02,844 INFO [main] o.eclipse.jetty.server.AbstractConnector Started ServerConnector@58516c91{HTTP/1.1,[http/1.1]}{0.0.0.0:8080}
2020-05-27 14:26:02,845 INFO [main] org.eclipse.jetty.server.Server Started @151563ms
2020-05-27 14:26:02,900 INFO [main] org.apache.nifi.nar.NarAutoLoader Starting NAR Auto-Loader for directory ./extensions ...
2020-05-27 14:26:02,901 INFO [main] org.apache.nifi.nar.NarAutoLoader NAR Auto-Loader started
2020-05-27 14:26:02,902 INFO [main] org.apache.nifi.web.server.JettyServer NiFi has started. The UI is available at the following URLs:
2020-05-27 14:26:02,903 INFO [main] org.apache.nifi.web.server.JettyServer http://******:8080/nifi
2020-05-27 14:26:02,903 INFO [main] org.apache.nifi.web.server.JettyServer http://127.0.0.1:8080/nifi
2020-05-27 14:26:02,906 INFO [main] org.apache.nifi.BootstrapListener Successfully initiated communication with Bootstrap
2020-05-27 14:26:02,907 INFO [main] org.apache.nifi.NiFi Controller initialization took 65344195582 nanoseconds (65 seconds).
2020-05-27 14:27:58,182 INFO [Write-Ahead Local State Provider Maintenance] org.wali.MinimalLockingWriteAheadLog org.wali.MinimalLockingWriteAheadLog@2407f1a8 checkpointed with 0 Records and 0 Swap Files i
n 27 milliseconds (Stop-the-world time = 4 milliseconds, Clear Edit Logs time = 4 millis), max Transaction ID -1

NIFI在手，跟我走！

在浏览器输入http://127.0.0.1:8080/nifi，进入Apache NIFI的交互界面。我们看到背景是网格式的，通常我习惯把它叫做设计页面，在这个设计页面上，我们可以通过拖拉拽的形式设计DataFlow。

1.准备表结构和数据

我在MySQL里新建了两张表，一个叫source来源表,一个叫target目标表。两张表的结构是一样的，其中increase设计成自动递增的，这样increase是一个增量字段。

增量字段顾名思义，数据库表里每次新来的数据的这个增量字段的值，都比上一次的大，严格意义上增量字段是递增且不重复的。(区别于将时间戳字段作为增量字段，通常业务里的时间戳字段都不是严格意义上的增量字段)

现在source表里还没有数据，这里我随意在NIFI里拉了两个组件往source表里写数据，你不用关心这里的处理，我只是在准备来源表的数据，你可以使用任何方式向source表里写数据。

最终我一共向source表里写了253001条数据。

2.新建一个Process Group

在NIFI交互界面的顶层，有一排的可拖拽按钮，按照如下图示，拖拽一个Process Group出来。

我们给这个Process Group起一个名字叫体验流程，在这里你可以简单的理解Process Group是一个组，在这个组里面可以设计你的流程，甚至设计其他的组。

双击体验流程这个Process Group，我们会进入一个新的空白的设计页面。

3.新建一个GenerateTableFetch组件

进入到体验流程这个Process Group后，如下图所示，我们左键点顶层第一个按钮，按住左键向空白页面拖拽，会弹出一个Add Processor的对话框。

在这个对话框里，我们找到GenerateTableFetch这个组件。

简单说一下GenerateTableFetch这个组件，它的作用就是根据指定的表和表字段(通常是一个增量字段)，生成一批SQL语句，这些SQL是分页的(或者说分片的)，这样一张有很多数据的一张表，我们就可以通过多个SQL分批的查询出来，这样会更高效。如果直接去全表扫描一张大表，有可能会等待很长时间，有可能会因为数据太多发生一些异常，这都不是我们想看到的。

双击GenerateTableFetch这个组件，这个组件就会出现在我们的设计页面上了。双击这个GenerateTableFetch组件或者鼠标右键选择Configure，会弹出Configure Processor对话框。

Configure Processor对话框有四个页签，这里我简单说明一下，不必纠结没有提及的那些配置究竟是什么意思。其他没有提及的配置你就当成是高级应用，不明白这些配置并不影响本文的阅读

SETTINGS：可以配置Terminate哪些RelationShip，Terminate你可以简单理解为我们忽略、不在意这个RelationShip。

RelationShip：每一个Processor处理完数据得到的结果，一般会将他们传输出去，RelationShip就是他们传输的方向。具体哪部分数据传输到哪个方向，由具体的程序代码决定。每个组件的RelationShip可能都不一样，具体的含义需要查看各个组件的详细说明文档，你可以右键组件，选择View usage来查看这个组件具体的用法。这里我们要注意一点就是，每个组件的所有RelationShip都应该有所指向(下面会提到将一个组件连接到另一个组件，组件中间会有一个Connection的东西，这个Connection会包含一个或多个RelationShip，那么这样的RelationShip就是有所指向的。另外Terminate也算是有所指向)

SCHEDUING：配置调度的地方，可用的调度的策略(Scheduling Strategy)有两种，一个是Timer driven,配置每多久执行一次调度。另一个是CRON driven,可以配置比如每天在几点几分执行一次调度。具体的调度时间是在Run Schedule里配置的。Concurrent Tasks是说一次调度，这个组件最多可以同时启动多少个任务来处理数据。Execution是针对集群的，你可以先不用理解，它是设置组件只在主节点运行还是在所有节点运行。

PROPERTIES：这个是每个组件的核心功能配置，每个组件的配置都是不一样的。

COMMENTS：注释，可以在里面添加一些描述信息。

4.配置GenerateTableFetch组件

SCHEDUING页签，我们选择Timer driven,Run Schedule配置成5 sec,即每5秒调度一次。

PROPERTIES页签，如下图，我们通过Database Connection Pooling Service新建了一个DBCPConnectionPool数据库连接池服务，Database Type选择了MySQL,Table Name配置填写了来源表名称source,Maximum-value Columns我们配置了增量字段increase

点击Apply后，回到设计页面，我们发现组件左上角是一个感叹号，感叹号你可以理解为这个组件还没有满足运行的条件，把鼠标停留在感叹号上会有提示信息

通过提示信息我们看到两个问题，第一个是DBCPConnectionPool数据库连接池服务还不可用，另一个是当前的组件success和failure两个RelationShip还没有指向。

5.配置DBCPConnectionPool服务

如下图所示，鼠标右键点击空白页面，选择Configure，进入Process Group的配置页面

进入Process Group的配置页面后，选择CONTROLLER SERVICES页签，我们可以看到我们之前建的DBCPConnectionPool数据库连接池服务。点击齿轮形状的配置按钮，如下图所示会弹出Configure Controller Service的对话框。

点击对话框的PROPERTIES页签，按如下图所示配置MySQL数据库的连接信息。

这里我们把本文的数据库连接列出来

jdbc:mysql://ip:port/nifi?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull&tinyInt1isBit=false

com.mysql.jdbc.Driver

驱动包需要自己去下载，然后添加到NIFI本机的某个地方
/data/nifi-1.11.4/jdbc/mysql-connector-java-5.1.46.jar

配置完毕后，点击APPLY，然后如下图，点击闪电符号按钮，启用DBCPConnectionPool数据库连接池服务。

6.配置GenerateTableFetch组件RelationShip

回到设计页面，我们看到GenerateTableFetch这个组件黄色感叹号的提示信息关于DBCPConnectionPool的已经没有了，现在提示的是关于RelationShip的。

现在我们通过新建GenerateTableFetch同样的方式，在设计页面新增一个ExecuteSQLRecord组件，然后将鼠标停留在GenerateTableFetch组件上，会出现一个箭头，点击拉取这个箭头然后指向ExecuteSQLRecord

指向的过程中会弹出Create Connection的配置页面，在里面的For Relationships勾选success(后期我们可以右键点击Connection进入配置页面)

点击ADD后，再次回到设计页面

我们看到GenerateTableFetch这个组件黄色感叹号的提示信息只剩下failure RelationShip的了。

此时我们再次到GenerateTableFetch的配置页面，在SETTINGS页面的Automatically Terminate Relationships里勾选failure。

点击APPLY后，GenerateTableFetch这个组件黄色感叹号就会消失了。

7.配置ExecuteSQLRecord组件

简单说一下ExecuteSQLRecord组件，执行上游传输过来的SQL语句，然后将查询结果以指定的数据格式输出到下游。

与配置GenerateTableFetch配置的操作流程都是大体相似的，这里不做重复性的叙述了，看过程图就可以了。

8.配置PutDatabaseRecord组件

我们在设计页面上新增一个PutDatabaseRecord组件，并做相应配置

简单说一下PutDatabaseRecord组件，以指定格式读取上游的数据，然后将数据insert/update/delete到指定的数据库表。

9.运行整个数据处理流程

右键点击每个组件选择start或者点击空白出选择start

可以看到已经有数据在流动被处理了。

写入数据总是比较慢的，这个时候我们可以适当的停止PutDatabaseRecord组件修改配置提高它的并发任务数(注意只有stop这个组件，才可以对它进行配置)

当我们再次运行PutDatabaseRecord组件，在设计页面会发现流程报错了(这并不是意外，这是我设计好展示给你们看的效果)

我们可以看到PutDatabaseRecord的右上角有一个8,并且GenerateTableFetch组件右上角有一个红色告警，将鼠标停留在红色告警，会有提示信息。看报错信息的意思是说GenerateTableFetch无法获取到数据库连接。(DBCPConnectionPool数据库连接池默认是8个连接，但这8个全被PutDatabaseRecord拿去用了，GenerateTableFetch拿不到连接所以报错了。)

好了，上面的报错设计只是为了让你看到多任务时组件右上角会有任务数的提示，当发生异常时组件的右上角也会有红色告警信息。