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/