NiFi-从部署到开发（图文详解）

NiFi简介

Apache NiFi 是一款强大的开源数据集成工具，旨在简化数据流的管理、传输和自动化。它提供了直观的用户界面和可视化工具，使用户能够轻松设计、控制和监控复杂的数据流程，NiFi 具备强大的扩展性和可靠性，可用于处理海量数据，并且能很好地应对复杂的数据转换需求，还可以设置定时调度任务

特点

• 可视化操作：提供了图形化界面，用户可以通过拖放组件来构建数据处理流程。
• 数据处理功能强大：能够实现数据的获取、转换、分发等操作。例如，可以从各种数据源获取数据，对数据进行格式转换、内容过滤等处理，再将数据发送到目标系统。
• 可靠性高：在数据传输和处理过程中具有良好的容错机制，保障数据的完整性和准确性。
• 扩展性好：可以轻松地扩展来处理大规模的数据流量和复杂的数据处理任务

NiFi的下载和安装

在本篇文章中主要讲述了四种部署方式，一种是单机部署比较简单，第二种是使用nifi自带的zookeeper部署伪分布模式,如果你的资源不足的话可以选用该模式，第三种是使用nifi自带的zookeeper部署集群模式，当然这种使用内部自带zookeeper的情况在实际开发中不常用，第三种就是比较常用的使用外部zookeeper的nifi集群模式的部署

首先我们先来看一下单机部署

单机部署

以下部署要在java环境下，若没有请事先安装jdk

下载安装包

进入Apache NiFi 官方网站来下载你所需要的版本

也可以直接点击nifi官网所有版本来下载

这里我下载的是1.13.2版本

下载完成后上传至Linux进行解压处理

解压：tar -zxvf  nifi-1.13.2-bin.tar.gz -C /opt/installs/

修改配置文件

cp nifi-1.13.2/conf/nifi.properties nifi-1.13.2/conf/nifi.properties.bak

修改ip和端口号

启动nifi

nifi-1.13.2/bin/nifi.sh

下面可以看到有启动/停止/运行/重启/状态等服务

启动nifi：nifi-1.13.2/bin/nifi.sh start

进入nifi的可视化界面

http://192.168.233.128:5800/nifi/

日志的位置

Nifi 集群部署-伪分布模式

由于nifi内置zookeeper，故我们先使用内置zookeeper进行搭建，可以在一台服务器上搭建伪分布模式，也可以在三台服务器上搭建集群模式，因为伪分布模式在一台机器上搭建，所以不同节点的相同功能端口会不同，如果搭建集群模式，IP不同，那么不同节点的相同功能端口可以相同，此处搭建集群模式

1、修改linux的⼀个安全机制

（1）进入vi /etc/selinux/config，添加SELINUX=disabled，防止后续出现一些问题

三台都需要修改，改完重启一下

（2）因为配置完单机模式后，nifi会产生许多新的数据库来存储数据，为了防止对伪分布模式有影响，这里先删除单机模式的nifi，重新解压一份

解压：tar -zxvf  nifi-1.13.2-bin.tar.gz -C /opt/installs/重命名：mv nifi-1.13.2/ nifi

rm -rf /opt/installs/nifi-1.13.2/

2、准备三个单机NIFI实例

nifi-1 nifi-2 nifi-3

3、修改配置文件

cp nifi/conf/nifi.properties nifi/conf/nifi.properties.bak

（1）修改三台服务器nifi中的zookeeper.properties

# 1节点2181，2节点2182，1节点2183clientPort=12181                                                                                                                    # 不同机器使用不同IPserver.1=bigdata01:12888:13888server.2=bigdata01:14888:15888server.3=bigdata01:16888:17888

（2）新建nifi-1/state/zookeeper，nifi-2/state/zookeeper'nifi-3/state/zookeeper在此文件夹中新建文件myid，分别对应写入1、2、3

（3）编辑节点conf/nifi.properties文件

1 ####################2 # State Management #                                                                                                 3 ####################4 nifi.state.management.configuration.file=./conf/state-management.xml                                             5 nifi.state.management.provider.local=local-provider  6 nifi.state.management.provider.cluster=zk-provider7 #  指定此NiFi实例是否应运行嵌入式ZooKeeper服务器，默认是false                          8 nifi.state.management.embedded.zookeeper.start=true                                                                9 nifi.state.management.embedded.zookeeper.properties=./conf/zookeeper.properties 
10 
11 # web properties #                                                 
12 nifi.web.war.directory=./lib    
13 # HTTP主机。默认为空白                                               
14 nifi.web.http.host=bigdata01
15 # HTTP端口。默认值为8080
16 nifi.web.http.port=18001
17 
18 # cluster node properties (only configure for cluster nodes) #   
19 # 如果实例是群集中的节点，请将此设置为true。默认值为false
20 nifi.cluster.is.node=true 
21 # 节点的完全限定地址。默认为空白
22 nifi.cluster.node.address=bigdata01
23 # 节点的协议端口。默认为空白
24 nifi.cluster.node.protocol.port=28001
25 
26 # 指定在选择Flow作为“正确”流之前等待的时间量。如果已投票的节点数等于nifi.cluster.flow.election.max.candidates属性指定的数量，则群集将不会等待这么长时间。默认值为5 mins
27 nifi.cluster.flow.election.max.wait.time=1 mins 
28 # 指定群集中所需的节点数，以便提前选择流。这允许群集中的节点避免在开始处理之前等待很长时间，如果我们至少达到群集中的此数量的节点
29 nifi.cluster.flow.election.max.candidates=1
30 
31 # cluster load balancing properties #  
32 nifi.cluster.load.balance.host=
33 nifi.cluster.load.balance.port=6342
34 
35 # zookeeper properties, used for cluster management # 
36 # 连接到Apache ZooKeeper所需的连接字符串。这是一个以逗号分隔的hostname：port对列表
37 nifi.zookeeper.connect.string=bigdata01:12181,bigdata01:12182,bigdata01:12183
38 nifi.zookeeper.connect.timeout=3 secs                                                      
39 nifi.zookeeper.session.timeout=3 secs                                                   
40 nifi.zookeeper.root.node=/nifi

bigdata01
143行：nifi.web.http.host=bigdata01
144行：nifi.web.http.port=18001
241行：nifi.cluster.is.node=true
242行：nifi.cluster.node.address=bigdata01
243行：nifi.cluster.node.protocol.port=28001
256行：nifi.cluster.load.balance.port=6342
262行：nifi.zookeeper.connect.string=bigdata01:12181,bigdata01:12182,bigdata01:12183

节点2,节点3内容跟节点1相同，只是nifi.web.http.port，nifi.cluster.node.protocol.port，nifi.cluster.load.balance.port，这三个端口区分开来，避免端口重复

（4）修改conf/state-management.xml文件

61行：<property name="Connect String">bigdata01:12181,bigdata01:12182,bigdata01:12183</property>

4、启动三个实例，启动完成后进入可视化界面

启动：nifi-1/bin/nifi.sh start
启动：nifi-2/bin/nifi.sh start
启动：nifi-3/bin/nifi.sh start
可视化界面：bigdata01:18001

Nifi 集群部署-内置Zookeeper

因为配置完单机模式后，nifi会产生许多新的数据库来存储数据，为了防止对集群模式有影响，这里先删除单机模式的nifi，重新解压一份，重命名为nifi

1、分发nifi至三台服务器

xsync nifi/

2、修改配置文件

（1）修改state-management.xml

<cluster-provider><id>zk-provider</id><class>org.apache.nifi.controller.state.providers.zookeeper.ZooKeeperStateProvider</class><property name="Connect String">bigdata01:2181,bigdata02:2181,bigdata03:2181</property><property name="Root Node">/nifi</property><property name="Session Timeout">10 seconds</property><property name="Access Control">Open</property></cluster-provider>

进行分发

xsync /opt/installs/nifi/conf/state-management.xml

（2）修改nifi.properties

bigdata01
51行：nifi.state.management.embedded.zookeeper.start=true（默认为false，如果使用外部的zookeeper集群为false）
143行：nifi.web.http.host=bigdata01
144行：nifi.web.http.port=18001
241行：nifi.cluster.is.node=true
143行：nifi.cluster.node.address=bigdata01
243行：nifi.cluster.node.protocol.port=28001
262行：nifi.zookeeper.connect.string=bigdata01:2181,bigdata02:2181,bigdata03:2181

进行分发并修改主机名

xsync /opt/installs/nifi/conf/nifi.properties
修改143行、143行主机名为相应的另外两台服务器的名字即可

（3）修改zookeeper.properties

clientPort=2181                                                                                                                    
server.1=bigdata01:2888:3888;2181
server.2=bigdata02:2888:3888;2181
server.3=bigdata03:2888:3888;2181

进行分发

xsync /opt/installs/nifi/conf/nifi.properties

（4）新建nifi/state/zookeeper，在此文件夹中新建文件myid，且在三台虚拟机中依次写入1、2、3

3、启动nifi并产看web界面

cd /opt/installs/nifibin/nifi.sh start #三台都要启动bigdata01:18001

Nifi 集群部署-外部Zookeeper

修改使用内置zookeeper状态下的nifi.properties中的第51行即可

web界面简介以及简单读取文件案例

选择自己想使用的处理器

填写文件夹路径（三台服务器都要有）

打勾了说明如果成功了就会停下来

把上述success打的勾取消，再拖拽入一个output类型的处理器进行连接，点击运行，发现getFile运行中，nifi的强大之处，只要有一个处理器能运行就会执行，此时我们点击运行，并往上述文件夹中发送数据，数据会先存储到管道中，数据会管道把数据存到本地磁盘，如果后续处理器发生错误也不会影响前面处理器的运行

点击小眼睛或者感叹号都可以查看内容

接下来配置输出路径，会发现数据传到了输出路径中，管道变成了0字节

从本地文件读取数据到MySQL

1、添加处理组file_to_mysql

开启hdfs服务：start-dfs.sh

2、填写hadoop中core-site-xml和hdfs-site.xml的路径

/opt/installs/hadoop/etc/hadoop/core-site.xml,/opt/installs/hadoop/etc/hadoop/hdfs-site.xml

其他参数可以先不配置

因为默认会加载你hadoop的配置文件

点击运行，往输入路径中写入文件

cp wcc.txt /home/data

离线同步MySQL数据到hdfs

类似于datax的功能

新建一个处理组mysql_to_hdfs

添加mysql端

添加驱动以及驱动地址

jdbc:mysql://bigdata01:3306/mydb01
com.mysql.jdbc.Driver
/opt/installs/hive/lib/mysql-connector-java-8.0.26.jar

这是因为没有填写账号密码

再次启动即可

这时发现文件传输速度特别快，可以修改调度时间，因为我们的离线数仓是一天调度一次，可以改成86400秒

查看hdfs的数据发现是乱码的

这时我们可以添加一个中间处理器将avro格式转换为json格式

添加之后再次启动即可

可以看到我们每次运行都把数据放入了同一个文件夹，这样是不行的，我们之前用datax导数据时放入了不同的文件夹

修改动态目录

添加dt后仍然无法识别，需要再添加一个处理器UpdateAttribute

给前三个处理器执行一下，使数据在第三个管道内

再运行第四个处理器，在hdfs上查看

但是还是要手动去写，能不能调用一些函数呢

双击可查看获取当前时间的方法

这里我们选用${now():format('yyyy')}并改成年月日的形式${now():format('yyyy-MM-dd')}

这样每天都会创建新的文件夹，但是打开文件夹后发现里面的文件名是用UUID命名的，虽然使文件名不重复但是不便于观看

以时间戳和后缀作为文件名，这样可以控制文件的滚动

再次运行可以看到文件名得到了修改

还可以修改成每个小时生成一个新文件

实时监控kafka数据到hdfs

启动kafka集群，添加kafka消费者，选择对应的版本

修改配置

复制一份puthdfs，修改输出路径

和上面一样通过添加日期函数使一个小时生成一个文件夹，这样可以解决小文件问题