Apache NiFi Processor实战

1 前言

Apache NiFi是什么？NiFi官网给出以下解释：“一个易用、强大、可靠的数据处理与分发系统”。通俗的来讲，即Apache NiFi 是一个易于使用、功能强大并且可靠的数据处理和分发系统，其为数据流设计，它支持高度可配置的指示图的数据路由、转换和系统中介逻辑。
为了对NiFi可以表述的更为清楚，下面经过NiFi的架构来作简要介绍，以下图所示。

根据官网对各个组件的说明，作摘要翻译：
WebServer：其目的在于提供基于HTTP的命令和控制API。
Flow Controller：这是操做的核心，以Processor为处理单元，提供了用于运行的扩展线程，并管理扩展接收资源时的调度。
Extensions：在其余文档中描述了各类类型的NiFi扩展，Extensions的关键在于扩展在JVM中操做和执行。
FlowFile Repository：FlowFile库的做用是NiFi跟踪记录当前在流中处于活动状态的给定流文件的状态，其实现是可插拔的，默认的方法是位于指定磁盘分区上的一个持久的写前日志。
Content Repository：Content库的做用是给定流文件的实际内容字节所在的位置，其实现也是可插拔的。默认的方法是一种相对简单的机制，即在文件系统中存储数据块。
Provenance Repository：Provenance库是全部源数据存储的地方，支持可插拔。默认实现是使用一个或多个物理磁盘卷，在每一个位置事件数据都是索引和可搜索的。

2 NiFi Processer介绍

上一节说了那么多，主要经过NiFi的架构图介绍了NiFi的基本概念，由概念可知Flow Controller是NiFi的核心，那么Flow Controller具体是什么？Flow Controller扮演者文件交流的处理器角色，维持着多个处理器的链接并管理各个Processer，Processer则是实际处理单元。那么，让咱们经过NiFi的UI看下NiFi的Processor包含哪些？

经过上图可知，Processor包含各类类型的组件，如amazon、attributes、hadoop等，可经过前缀进行轻易辨识，如Get、Fetch开头表明获取，如getFile、getFTP、FetchHDFS，execute表明执行，如ExecuteSQL、ExecuteProcess、ExecuteFlumeSink等都可较容易知其简单用途。

3 NiFi Processer实战

说了那么多，介绍了NiFi的架构和Processor，那么说好的实战呢？那么，本文就以笔者的一个实际需求为例，进行Processor的实战。需求以下：选取一款数据处理调度工具，对服务器脚本实现定制调度执行。其中服务器的脚本涉及到对环境变量、oracle数据库、Hadoop生态圈组件的调度。当对服务器脚本调度执行完成后返回脚本运行状态，并提供失败重运行接口。
为了实现需求，曾调度过各类调度工具，如Apache Oozie、Azkaban、Pentaho等，最终比较了各类利弊尝试选用Apache NiFi做为尝试，经过查阅NiFi Processor API，能更好的支持远程操做的Processor为ExecuteProcess。下面将对需求进行实战讲解。

3.1 Processor的添加与配置

1. 点击“Add Processor”，选择ExecuteProcess后点击Add按钮完成添加，以下图。

2. 右击ExecuteProcess后选择Configure Processor，对Properties选项卡进行配置，其中每个配置选项均提供了相关的说明，以下图。

如上图所示，这里有必要对各选项进行相关说明。
Command（执行命令）： sh。
Command Arguments（执行命令参数）：-c;ssh user@ip sh js/job/job_hourly.sh `date
Batch Duration（执行间隔时间）：不设置。//咱们需求是经过定时调度，而并不是按间隔时间执行。
Redirect Error Stream（重定位流）：不设置。
Argument Delimiter（执行命令参数分隔符）：; //以；对参数进行分割。

3.2 Processor调度

NiFi支持三种调度策略，包括Time Driven（时间驱动）、CRON Driven（CRON驱动）和Event Driven（事件驱动，非可选），根据咱们实际需求选择CRON Driven，我的理解CRON便是Crontab的应用，CRON的各参数含义分别表明：秒、分、时、日、月、周、年，须要配合*、？和L共同执行（*表明字段的值都有效；?表明对于指定的字段不指定值；L表明长×××）。如：“0 0 13 * * ?”表明想要在天天下午1点进行调度执行。所以根据咱们的需求进行参数的调度配置。以下图所示。

3.3 运行状态监控

NiFi经过Rest API供开发者调度，这里咱们用Processor API对运行状态进行监控（状态参数获取、Processor的启动与中止）。
1. 运行状态监控参数获取：
命令以下：curl ‘http://IP/nifi-api/processors/processorsID ‘获得以下结果，可经过json解析器解析并获取状态。

2. Processor的启动与中止：
NiFi的Processor启动中止经过其Put方法实现，Put最有效的做用是改变其运行状态，NiFi的Process总共有三种状态，即Running、Stopped和Disabled。
那么咱们将开始和中止两个命令Rest API的放在脚本中执行便可。
启动命令（使用Rest API的Put方法）：
curl -i -X PUT -H ‘Content-Type:application/json’ -d ‘
{
“revision”: {
“clientId”: “586ec1d7-015d-1000-6459-28251212434e”,
“version”:17},
“component”: {
“id”: “39e0dafc-015d-1000-918d-bee89ae2226e”,
“state”: “RUNNING”
}
}’ http://IP/nifi-api/processors/processorsID
中止命令（使用Rest API的Put方法）：
curl -i -X PUT -H ‘Content-Type:application/json’ -d ‘
{
“revision”: {
“clientId”: “586ec1d7-015d-1000-6459-28251212434e”,
“version”:17},
“component”: {
“id”: “39e0dafc-015d-1000-918d-bee89ae2226e”,
“state”: “STOPPED”
}
}’ http://IP/nifi-api/processors/processorsID

4 小结与后记

本文首先对Apache NiFi进行简介，后以笔者的实际需求为例，对NiFi核心组件Processor的实战说明。因为NiFi仍然属于Apache推出时间不长的一个顶级项目，虽功能十分强大，但可查阅资源仍然有限，本文更多的是一个抛砖的过程，其真正强大的功能还在数据处理上，欢迎感兴趣的各位进行互相探讨。

推荐你们一个开源私有化部署的用户行为分析平台Cobub Razor，项目地址：https://github.com/cobub/razor

文／hPan  原文连接：http://www.cobub.com/actual-combat-of-apache-nifi-processor/