✨Nifi系列✨ Nifi同步并解析FTP服务器Excel文件，并存储到数据库

一、GetFTP功能介绍

使用GetFTP Processor去FTP服务器中同步文件，通过修改GetFTP配置项，实现不同的功能。

✨Nifi系列✨-GetFTP功能介绍，FTP服务器数据采集

1.GetFTP基础配置项

SCHEDULING

如果Nifi部署的分布式集群的话，建议使用Primary node模式，否则每个节点都会从FTP服务器读取一次文件，造成数据重复。

至于调度时间，根据自己实际需求而定。

PROPERTIES

FTP服务器的IP、端口、用户名、密码为必填项

启动

通过简单修改调度项，以及配置FTP服务器的基础信息后，就已经可以成功同步到数据文件

2.Delete Original 配置项

默认情况下，Delete Original 为true，也就是每次同步成功后，都会把FTP服务器上的源数据删除。但是很多情况下，我们只是希望同步数据，并且保留FTP服务器源数据文件。

当Delete Original 为true时，同步成功后，删除源数据文件;

当Delete Original 为false时，同步成功后，不删除源数据文件;

Delete Original = true

在FTP服务器中创建一个名为zxy_delete_original的文件

重新启动后，通过list queue查看文件已经读取到

与此同时，FTP服务器中该文件已经被删除

Delete Original = false

在FTP服务器中重新创建一个名为zxy_delete_original的文件。

重新启动后，通过list queue查看文件已经读取到

再次检查FTP服务器中源文件，仍然存在

3.Search Recursively

通过修改配置项Search Recursively,决定在同步FTP服务器的时候，是否递归搜索。默认为false。

当Search Recursively为true时，递归搜索；

当Search Recursively为false时，非递归搜索；

测试文件如下，且文件夹b下也有文件。

Search Recursively = true

设置递归搜索时，会去检索FTP服务器下所有文件

Search Recursively = false

设置非递归搜索时，只能获取到根目录下的文件，而不能检索根目录文件里的其他文件

4.Use UTF-8 Encoding

是否使用UTF-8编码

当Use UTF-8 Encoding为true时，使用UTF-8编码；

当Use UTF-8 Encoding为false时，不使用UTF-8编码；

测试文件如下，创建一个中文表名，查看不同配置下的同步情况。

Use UTF-8 Encoding = true

当设置UTF-8编码的时候，读取到的中文表名称是正常的。

Use UTF-8 Encoding = false

当设置UTF-8编码的时候，读取到的中文表名称是乱码的。

二、Nifi同步并解析FTP服务器Excel文件，并存储到数据库

FTP服务器中存有Excel文件，希望将Excel文件中数据转储到数据库，在Nifi中使用GetFTP -> ConverttExcelToCSVProcessor -> UpdateAttribute -> ConvertRecord -> SplitJson -> EvaluateJsonPath -> ReplaceText -> PutSQL流程。

1.Excel数据同步

1.1 待同步数据

模拟当前FTP服务器中有一个Excel文件zxy.xlsx，文件内数据如下:

根据已知数据在数据库中建表:

create table zxy.province_and_city(
province varchar(25) null,
city varchar(25) null,
area varchar(25) null
)
1
2
3
4
5

1.2 GetFTP

GetFTP基础配置介绍

配置FTP服务器的基础信息，以及配置递归搜索、不删除源数据文件、支持UTF-8编码

1.3 ConvertExcelToCSVProcessor

将读取到的Excel转换成CSVProcessor，主要配置在于CSVFormat为Microsoft Excel

1.4 UpdateAttribute

默认即可

1.5 ConvertRecord

配置Record Reader和Record Writer，将CSV转换成JSON类型

Record Reader需要配置CSVReader，如果没有这个选项，通过Create new service创建即可，然后启动CSVReader

Record Writer需要配置JsonRecordSetWriter，同上

当配置好这两项后，正确结果应该如下:

1.6 SplitJson

默认即可

1.7 EvaluateJsonPath

在EvaluateJsonPath中，需要获取JSON中的每个字段的数据并重命名。Destiation设置为flowfile-attribute，即将area、city、province分发到后面的流程中。

1.8 ReplaceText

通过上一步获得了各个字段名称，可通过${}引用值，在Replacement Value处预写SQL。

insert into province_and_city values('${province}','${city}','${area}')

1.9 PutSQL

只需要配置JDBC Connection Pool即可

1.10 启动流程

select *
from zxy.province_and_city;
1
2

2.多Excel数据同步

如果有两个表结构一样的Excel文件，那么我将数据存储到数据库后，我怎么区分数据原来属于哪个excel呢？

2.1 数据库修改

# 在数据库的zxy.province_and_city表中添加filename字段，用于存储excel文件名
alter table zxy.province_and_city add column filename varchar(25) null ;

1
2
3

2.2 流程修改

整个流程中，只需要修改ReplaceText Processor的Replacement Value项即可。

修改为insert into province_and_city values('${province}','${city}','${area}','${filename}')

2.3 启动流程

修改完ReplaceText后，重启启动整体流程即可。

同步完成后，通过查询数据库，可以看出已经将filename插入到数据库中。filename的组成是Excel表名_Sheet页名.csv