NiFi 集群安装部署及使用-蒲公英云

NiFi 架构

NiFi在主机操作系统上的JVM内执行。JVM上NiFi的主要组件如下：
1. Web Server
  Web服务器的目的是托管NiFi基于HTTP的命令和控制API。
2. Flow Controller
  流量控制器是操作的大脑。它为扩展程序提供运行的线程，并管理扩展程序何时接收要执行的资源的计划。
3. Extensions
  在其他文献中描述了各种类型的NiFi扩展。这里的关键点是扩展在JVM中运行和执行。
4. FlowFile Repository
  FlowFile存储库是NiFi跟踪其对流中当前活动的给定FlowFile的了解状态的地方。存储库的实现是可插入的。默认方法是位于指定磁盘分区上的持久性预写日志。
5. Content Repository
  内容存储库是给定FlowFile的实际内容字节。存储库的实现是可插入的。默认方法是一种相当简单的机制，它将数据块存储在文件系统中。可以指定多个文件系统存储位置，以便获得不同的物理分区以减少任何单个卷上的争用。
6. Provenance Repository
  Provenance Repository是存储所有出处事件数据的地方。存储库构造是可插入的，默认实现是使用一个或多个物理磁盘卷。在每个位置内，事件数据被索引并可搜索。
下载 NiFi 安装包并解压
下载地址: http://nifi.apache.org/download.html

单机部署

修改 con/nifi.properties 配置文件(可不修改)

# HTTP 主机地址。默认为空。
nifi.web.http.host=DSJ-TVM001
# HTTP 端口。默认为8080
nifi.web.http.port=8080

启动 NiFi

在 ${NIFI_HOME}/bin 目录中，执行命令./nifi.sh .

command	描述
start	后台启动 NiFi
stop	停止后台运行的 NiFi
status	查询 NiFi 的当前状态
run	在前台启动 NiFi，Ctrl+C 关闭 NiFi
install	将 NiFi 安装为服务。使用 service nifi {start\|stop\|run\|restart\|status}

访问地址: http://localhost:8080

集群部署

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzIxNTI1MA_size_16_color_FFFFFF_t_70 2
使用 NiFi 内嵌 ZooKeeper
集群的默认状态提供程序为 ZooKeeperStateProvider。这意味着 NiFi 依赖ZooKeeper才能表现为群集。在许多部署 NiFi 的环境中，可能没有集成 ZooKeeper 集群。为避免必须单独维护的 ZooKeeper 集群，NiFi 提供了启动嵌入式 ZooKeeper 集群。通常，建议在3个或5个节点上运行ZooKeeper。
参考: http://nifi.apache.org/docs/nifi-docs/html/administration-guide.html\#clustering

修改配置文件 nifi.properties

# 当前 NiFi 实例是否运行嵌入式ZooKeeper服务器
nifi.state.management.embedded.zookeeper.start=true
# ZooKeeper 属性文件
nifi.state.management.embedded.zookeeper.properties=./conf/zookeeper.properties
# 当前 nifi 节点主机名
nifi.web.http.host=cdh01
#nifi.web.http.host=cdh02
#nifi.web.http.host=cdh03
# 当前 nifi 节点端口
nifi.web.http.port=8990
# 是否是集群中的节点，默认值为false。
nifi.cluster.is.node=true
# 设置为当前节点的主机名
nifi.cluster.node.address=cdh01
#nifi.cluster.node.address=cdh02
#nifi.cluster.node.address=cdh03
# 此端口设置为高于1024的开放端口(任何较低的端口都需要root)。
nifi.cluster.node.protocol.port=8888
# 用于与集群中其他节点通信的线程数。此属性默认为10。
nifi.cluster.node.protocol.threads=10
#与集群中其他节点进行通信的最大线程数。此属性默认为50。
nifi.cluster.node.protocol.max.threads=50
# 指定在选择流作为“正确”流之前要等待的时间
nifi.cluster.flow.election.max.wait.time=5 mins
# 指定集群中导致流的早期选择所需的节点数
nifi.cluster.flow.election.max.candidates=1
# cluster 负载均衡配置 #
nifi.cluster.load.balance.host=cdh01
nifi.cluster.load.balance.port=6342
nifi.cluster.load.balance.connections.per.node=4
nifi.cluster.load.balance.max.thread.count=8
nifi.cluster.load.balance.comms.timeout=30 sec
# ZooKeeper 列表 例如: cdh01:2181,cdh02:2181,cdh03:2181。
nifi.zookeeper.connect.string=cdh01:2181,cdh02:2181,cdh03:2181
nifi.zookeeper.connect.timeout=3 secs
nifi.zookeeper.session.timeout=3 secs
# ZooKeeper 存储数据根目录。默认值为 /nifi
nifi.zookeeper.root.node=/nifi

修改 conf/state-management.xml 配置文件

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

修改 conf/zookeeper.properties 配置文件

initLimit=10
autopurge.purgeInterval=24
syncLimit=5
tickTime=2000
dataDir=./state/zookeeper
autopurge.snapRetainCount=30
server.1=cdh013888;2181
server.2=cdh023888;2181
server.3=cdh033888;2181

创建 ZooKeeper 节点ID

#第一个节点
cd $NIFI_HOME
mkdir state
mkdir state/zookeeper
echo 1 > state/zookeeper/myid
#第二个节点
cd $NIFI_HOME
mkdir state
mkdir state/zookeeper
echo 2 > state/zookeeper/myid
#第三个节点
cd $NIFI_HOME
mkdir state
mkdir state/zookeeper
echo 3 > state/zookeeper/myid

参考: http://nifi.apache.org/docs/nifi-docs/html/administration-guide.html\#embedded\_zookeeper

各个上节点启动 NiFi
NiFi 集群中的任一节点均可访问

NiFi 示例
1. NiFi 首页
2. 添加处理器
3. 配置处理器(GetFile 和 PutFile)
  GetFile
  
  配置说明:
  
  参考: http://nifi.apache.org/docs.html
  
  PutFile
4. 连接处理器
5. 启动和停止处理器
  启动:
  
  停止: