NiFi文檔

1.初識NiFi

1.1.概述

NiFi最早是美國國家安全局內部使用的工具,用來投遞海量的傳感器數據,后來由apache基金會開

源。NiFi基本設計理念：FlowBasedProgramming,

1.2.核心概念

?FlowFile

FlowFile表示在系統中移動的每個對象，FlowFile由兩部分組成：

ocontent內容，即數據本身

oattributes屬性，每條數據帶上的屬性信息.以鍵值對的形式.

?FlowFileProcessor

oFlowFile處理器，由它完成對數據的實際處理工作,包括但不限于對數據內容和屬性的加載，路

由，轉換，輸出等.

o處理器最靈活之處在于處理器可以讀寫FlowFile的屬性信息,并且用自帶的領域特定語言(DSL)

對屬性進行編程.

?Connection

。由Connections把各個處理器鏈接起來，從而形成數據處理流程的有向無環圖(DAG圖).也稱

數據流，NiFi中的Flow.

。Connection同時充當處理器間的隊列，并且隊列的屬性高度可配置.

o這些隊列可以配置優先級，可以在設置閾值，可以實現反壓。

?FlowController

o流控制器對用戶不可見的,它充當維護處理器如何連接和管理所有處理器所使用的線程及其分

配的重要角色。

FlowController充當促進處理器之間FlowFiles交換的代理。

ProcessGroup

o為了方便管理，把一組特定的處理器及其連接組成的Flow放到一個處理組中去，可以通過輸

入端口接收數據并通過輸出端口發送數據。

o以這種方式，處理組可以通過組合其他組來創建全新組，形成更加復雜的DAG圖（Flow流）。

1.3.關鍵特性

?Flow流高度可管理

?保證交付

NiFi的一個核心理念是即使在非常高的規模下，保證交付也是必須的。這是通過有效使用專門的

持久化的預寫日志（WAL）和內容存儲庫來實現的。它們的設計可以實現非常高的事務處理速率，

有效的負載分散，寫入時復制以及發揮傳統磁盤讀/寫的優勢。

?背壓和數據緩沖機制

NiFi支持緩沖所有隊列數據，以及在這些隊列達到指定限制時提供背壓的能力，或者在數據達到

指定時間時使數據過期失效。

?可配置優先級的隊列

NiFi允許設置一個或多個優先級策略，用于如何從隊列中檢索數據。默認是先進先出，但有時候

應該先拉取最新的數據，最大的數據或其他一些自定義方案。

?Flow流可配置特定的QoS（延遲v吞吐量，容量損失等）

在Flow流中有一些點是很關鍵的，且不能容忍丟失.或者有時候必須在幾秒鐘內處理和交付它。

NiFi可以對這些問題進行細粒度的特定配置。

?易于使用

?可視化的控制和命令

得益于強大的web操作界面.無論多么復雜的數據流都能在web界面上直觀的呈現,整個數據

處理流程，包括設計，控制，反饋和監控都可在web界面完成，一步到位.任何更改都能在界面上立馬

生效,完全不要部署的過程,對于整個數據流，更可以對中間某個處理器進行單獨變更，實時生效.

?數據流模板

對于設計好的數據流處理流，可以保存為模板來進行復用.模板可以導出成xml文件，導入到其他

NiFi中進行多處使用.

?數據溯源

flowfile流過Flow流時，NiFi會自動記錄，索引并提供可用的起源數據，包括導入，導出，轉

換等。這些信息對于故障排除，優化等很有用處.

?對歷史數據進行細粒度的恢復

NiFi的內容存儲庫旨在充當歷史記錄的滾動緩沖區。數據僅在內容存儲庫過期時或存儲空間不足

時才會被刪除。這與數據起源能力相結合，提供了非常精細的操作功能.包括對數據歷史中的某一

個點的點擊查看內容，下載內容，處理回放等功能.所有數據都可以回溯到它生命周期中很早的某一

點.

安全機制

?系統內部安全

Flow流中的流動的數據都可以進行加密傳輸

?用戶使用安全

支持用戶認證和不同級別的用戶授權（可讀，管理數據流，系統管理）

?多租戶授權

?可擴展的架構設計

?可擴展組件

NiFi的核心設計就是擴展.它的processors,ControllerServices,ReportingTasks,Pri

oritizers,andCustomerUserInterfaces都是可擴展的.

?隔離的類加載器

自定義的類加載器保證了擴展的組件簡單的依賴關系.

?點到點的通信協議

NiFi實例之間的通信協議是NiFi點到點（S2S）協議。S2s可以輕松，高效，安全地將數據從

一個NiFi實例傳輸到另一個實例。NiFi客戶端的庫也可以輕松在其他應用程序使用，以通過S

2s來與NiFi實例進行通信。S2s中支持基于套接字的協議和HTTP（S）協議作為底層傳輸協

議，使得可以將代理服務器嵌入到S2s通信中。

?靈活的擴容模型

?更多的NiFi實例

可以搭建NiFi集群，也可以不組成集群，多臺機器使用點到點協議來協作.

?更大的并發數量

直接修改處理器的并發數

1.4.架構

匚OS/Host

白JVMQWebServer

芯FlowController

Processor1ExtensionN

mFlowFile塞Content塞Provenance

RepositoryRepositoryRepository

LocalStorage

?WebServer

web服務器的提供基于http的命令和控制API。

?FlowController

流量控制器是操作的大腦。它為擴展程序提供運行所需的線程，并管理擴展程序何時接收執行資源的

時間表。

?Processor

處理組件

?Extensions

擴展組件

?FlowFileRepository

通過FlowFileRespository可跟蹤Flow中處于活動狀態的FlowFile的狀態。存儲庫的實現是插件

式的，默認是位于指定磁盤分區上的持久性預寫日志。

?ContentRepository

ContentRepository作為FlowFile的存儲庫，實現是插件式的，默認是一種相當簡單的機制，該

機制將數據塊存儲在文件系統中。可以指定多個文件系統存儲位置，以便使用不同的物理分區以減少任何

單個卷上的爭用。

?ProvenanceRepository

ProvenanceRepository是存儲所有來源事件數據的地方。存儲庫實現是插件式的，默認實現是使

用一個或多個物理磁盤卷。在每個位置內，事件數據都被索引并可以搜索。

集群

QOS/Hosto

aQOS/HosI

不8、QOS/Hosto

戈S、OS/Hostoo

ZooKeeperServer

?ClusterCoordinator

?PrimaryNode

?ZooKeeperClient

從NiFi1.0版本開始，采用了零主群集的范例。NiFi群集中的每個節點都對數據執行相同的任務，

但是每個節點都對不同的數據集進行操作。通過ZooKeeper選擇一個節點作為集群協調器，并且故障轉

移由ZooKeeper自動處理。所有群集節點均向群集協調器報告心跳和狀態信息。群集協調器負責斷開和

連接節點。此外，每個群集都有一個主節點，該節點也由ZooKeeper選擇。作為DataFlow管理者，您

可以通過任何節點的用戶界面（UI）與NiFi群集進行交互。您所做的任何更改都將復制到群集中的所有節

點，從而允許多個入口點。

2.源代碼淺析

2.1.1.總體結構

▼?Snifi-1.10.0[nifi]-/SRC/nifi-1.10.0<modules>

?■.github<module>nifi-commons</module>

?■.idea<module>nifi-api</module>

?nifi-api<module>nifi-framework-api</module>

Anifi-assembly<module>nifi-bootstrap</module>

?nifi-bootstrap

<module>nifi-mock</module>

?*2nifi-commons

<module>nifi-nar-bundles</module>

A■"nifi-docker

<module>nifi-assembly</module>

?nifi-docs

AB"nifi-external<module>nifi-docs</module>

?nifi-framework-api<module>nifi-maven-archetypes</module>

?nifi-maven>archetypes<module>nifi-external</module>

?nifi-mock<module>nifi-toolkit</module>

a*2nifi-nar-bundles<module>nifi-docker</module>

AB"nifi-toolkit</modules>

?nifi-api

就是nifi的應用程序接口，里面就是定義了整個工程用到的接口，注解，抽象類和枚舉等基本的接口

和信息.

?nifi-assembl

負責nifi的成品裝配，工程打包最后形成的可供部署的壓縮包就在這個工程里的target目錄內.

?nifi-bootstarp

負責nifi這個jvm應用程序的啟動相關事宜

?nifi-commons

nifi諸多特性，比如datd-provenance,expression-language.s2s傳輸的實現就在這里,同時也

是nifi的工具類集合

?nifi-docker

nifi的docker應用相關

?nifi-docs

nifi的文檔實現相關

?nifi-external

nifi內部元信息和外部交換,主要用于集群模式下

?nifi-framework-api

這就是nifi核心框架層的api,也就是架構圖中的FlowController那一層，注意這里只是各種接口

信息定義，不是實現.

?nifi-maven-archetypes

這里只是為了生成兩個mavenarchetype,一個是nifi自定義處理器的腳手架,一個是nifi自定

義服務的腳手架,這些腳手架在maven的中央倉庫都有提供.

?nifi-mock

用于nifi的mock測試

?nifi-nar-bundles

nifijava工具箱就是這里了.整個nifi里面大部分的maven工程都是這個工程的子工程,在這個

工程里面，一個bundle就是一個工具，也對應著上面架構圖里的Extension

?nifi-toolkit

這里面是nifi的命令行工具的實現.nifi也提供了比較豐富的命令行指令.

2.1.2.Nifi程序入口

在nifi-bpootstrap模塊內有一個org.apache.nifi.bootstrap.RunNifi的類，該類的main。方法

即為Nifi的啟動入口方法。

▼Bjnifi-bootstrapSystem.ouf.println(,

System.out.println("Restarx:StopApacheN1FJ,:fitisrunning,andthensti

▼?src

Systea.out.println("5tatus：DetermineifthereisarunningInstanceofApact

▼feimain

Systm.our.printIn("Dump:WriteaThreadDuaptothefilespecifiedby[optic

▼BijavaSystea.out.println(Diagnostics:Writediagnosticinformationtothefilesp<

▼Biorg.apache.nifi.bootstrap"thefilename,whichaoyresultinadditionaldiagnosticinfor?ationbeinc

?BiexceptionSyste?.our.println(°Run:StartanewinstanceofApacheNiFiandMonitorthe

?BinotificationSystefl.out.prlntlnO;

?ESutil>

?GBootstrapCodec

?QNIFILIstener@privatestaticString!)■>■(finalString!)orig)(returnArrays.copyOfRange(or:

1▼0RunNiFipublicstaticvoidmain(String[]args)throwslOExceptlon,InterruptedException(1

.if〔aros.lengthv111lengtha3)41

SfirtusprintUsagel];

ORunNiFi(File)return;

QcreateSensitiveKeyFile(File):Pat}

Gdiagnostics(Flle,boolean):void

Odump(File)：voidFiledunpEile-null;

Oenv():voidbooleanverbose■false：

QgetBootstrapFlle(Logger,String

finalStringand=args(0];

0getChildProcesses(String):List<

If(cnd.equal.5lgnoreCase(?nothe?Stnng："dump")){

QgetCurrentPort(Logger)：lnteger

if(argselength>1){

0getDefaultBootstrapConfFile():FduapFlle-newFHe(arg$(l]);

QgetFHe(String,File):File}else(

QgetHostname():String(MpfXte■null：

GqetLockFile(Loqqer):Fil

接著看start方法，里面做了很多前期的準備性工作，主要是加載bootstrap.conf里配置的屬性，以及在

里面構建另外一個javacmd命令：

finalProcessBuilderbuilder=newProcessBuilder():

mand(cmd)

Processprocess=builder.startf);

所以這個start方法是啟動了另外一個java進程,這個進程才是真正的NiFiruntime。

通過代碼跟蹤或查看日志，可見，這個cmd命令類似如下：

/opt/jdkl.8.0_131/bin/java

-classpath

/opt/nifi-1.7.1/./conf:/opt/nifi-1.7.1/./lib/logback-core-1.2.3.jar:/opt/nifi-1.7.1/./lib/jetty-schemas-3.1.jar:/o

pt/nifi-1.7.1/./lib/logback-classic-1.2.3.jar:/opt/nifi-1.7.1/./lib/jul-to-slf4j-1.7.25.jar:/opt/nifi-1.7.1/./lib/jcl-o

ver-slf4j-1.7.25.jar:/opt/nifi-1.7.1/./lib/nifi-properties-1.7.1.jar:/opt/nifi-1.7.1/./lib/nifi-runtime-1.7.1.jar:/o

pt/nifi-1.7.1/./lib/nifi-framework-api-1.7.1.jar:/opt/nifi-1.7.1/./lib/nifi-nar-utils-1.7.1.jar:/opt/nifi-1.7.1/./lib/

javax.servlet-api-3.1.0.jar:/opt/nifi-1.7.1/./lib/log4j-over-slf4j-1.7.25.jar:/opt/nifi-1.7.1/./lib/slf4j-api-1.7.25

.jar:/opt/nifi-1.7.1/./lib/nifi-api-1.7.1.jar

-Dpiler.disablejsrl99=true

-Xmx3g-Xms3g

-Djavax.security.auth.useSubjectCredsOnly=true

-Djava.security.egd=file:/dev/urandom

-D.http.allowRestrictedHeaders=true

-D.preferIPv4Stack=true

-Djava.awt.headless=true-XX:+UseGlGC

-Dtocol,handler.pkgs=.tocol

-Duser.timezone=Asia/Shanghai

-Dperties.file.path=/opt/nifi-l.7.l/./conf/perties

-Dnifi.bootstrap.listen.port=56653

-Dapp=NiFi

以盯礪日o「g.apache.nifi.NiFi，

可以清晰的看到，命令中實際執行的是java類org.apache.nifi.NiFi的main方法。

2.1.3.Nifi啟動初始化

這個org.apache.nifi.NiFi類在以下模塊中：

nifi-nar-bundles

nifi-framework-bundle

+---nifi-framework

+---nifi-runtime

Nifi-framework模塊就是nifi框架的核心代碼

Org.apache.nifi.NiFi.main。方法如下:

Mainentrypointoftheapplication.

*@paramargsthingswhichareignored

7

publicstaticvoidmain(String[]args){

LOGGERinfo("LaunchingNiFi...");

"y{

NiFiPropertiesproperties=convertArgumentsToValidatedNiFiProperties(args);

newNiFi(properties);

}catch(finalThrowablet){

LOGGERerror("FailuretolaunchNiFidueto"+1.1);

)

)

Main。方法調用了NiFi的構造方法:

publicNiFi(finalNiFiPropertiesproperties)

throwsClassNotFoundException,lOException,NoSuchMethodException,InstantiationException,

HlegalAccessException,IHegalArgumentException.InvocationTargetException{

this(properties.ClassLoader.getSystemClassLoader());

}

publicNiFi(finalNiFiPropertiesproperties.ClassLoaderrootClassLoader)

throwsClassNotFoundException.lOException,NoSuchMethodException.InstantiationException,

HlegalAccessException,IHegalArgumentException,InvocationTargetException{

第二個構造方法是實際上的構造方法，里面進行了大量初始化操作，以下是非常關鍵的部分:

//expandthenars

finalExtensionMappingextensionMapping=NarUnpacker.unpack/Vars(properties,systemBundle);

//loadtheextensionsclassloaders

NarClassLoadersnarClassLoaders=NarClassLoadersHolder.gef/nstence():

narClassLoaders.init(rootClassLoader,

properties.getFrameworkWorkingDirectoryO,properties.getExtensionsWorkingDirectoryO);

這部分初始化了NiFi的Java獷展工具箱，這些工具從Nifi的用戶來說就是在NiFi安裝目錄的lib目錄下

的各個*.nar包，這個些nar包實際就是NiFi增加了特定額外信息的jar包集合的壓縮，本質上還是jar

包。以下是我們解壓開的一個nar包，結構如下：

▼toMETA-INF

▼■maven

▼■org.apache.nifi

▼■nifi-pol-nar

Bperties

過pom.xml

▼■bundled-dependencies

Abcpkix-jdk15on-1.60.jar

Rbcprov-jdk15on-1.60.jar

0commons-codec-1.10.jar

Rcommons-collections4-4.2.jar

?commons-compress-1.18.jar

Acommons-csv-1.5.jar

Rconrwnons-io-2.6.jar

Acommons-lang3-3.8.1.jar

啟commons-text-1.4.jar

篇curvesapi-1.04.jar

Rjack$on-annotations-2.9.0.jar

口jackson-core-2.9.7.jar

Rjackson-databind-2.9.7.jar

啟jBCrypt-0.4.1.jar

Rnffi-poi-processors-1.9.2.jar

Anifi-processor-utils-1.9.2.jar

國nifi-security-utils-1.9.2.jar

Anifi-standard-record-utils-1.9.2.jar

Anifi-utils-1.9.2.jar

0poi-4.0.0.jar

Apoi-ooxml-4.0.0.jar

Rpoi-ooxml-schemas-4.0.0.jar

0xerceslmpl-2.11.0.jar

啟xml-apis-1.4.01jar

Rxmlbeans-3.0.1.jar

■DEPENDENCIES

■LICENSE

■NOTICE

■MANIFEST.MF

那么回到NiFi的構造方法內，首先是解壓這些nar包,并在代碼內用ExtensionMapping對象描述,

代碼如下：

//expandthenars

finalExtensionMappingextensionMapping=NarUnpacker.unpackA/ars(properties.systemBundle);

然后加載并初始化這些類加載器:

//loadtheextensionsclassloaders

NarClassLoadersnarClassLoaders=NarClassLoadersHolder.^ef/nstance(),

narClassLoaders.init(rootClassLoader,

properties.getFrameworkWorkingDirectory(),properties.getExtensionsWorkingDirectoryO);

//loadtheframeworkclassloader

finalClassLoaderframeworkClassLoader=narClassLoaders.getFrameworkBundle().getClassLoader();

if(frameworkClassLoader==null){

thrownewHlegalStateException("UnabletofindtheframeworkNARClassLoader.")：

)

finalSet<Bundle>narBundles=narClassLoaders.getBundles():

在NiFi的官方介紹中，有兩處它的特性介紹是擴展和類加載隔離,這里我們可以對它這兩個特性的實

現?探究竟了.它為每一個nar包構造了一個獨立的自定義的類加載器：NarClassLoader

publicclassNarClassLoaderextendsURLCIassLoader{

privatestaticfinalLoggerLOGGER=LoggerFactory.getLogger(NarClassLoader.class);

privatestaticfinalFileFilterJAR_FILTER-newFileFilter(){

@Override

publicbooleanaccept(Filepathname){

finalStringnameToTest=pathname.getName().toLowerCase();

returnnameToTest.endsWith(".jar")&&pathname.isFile();

}

}；

目前基本清晰，NiFi的擴展性是由自定義的壓縮文件nar包和自定義的類加載器來提供的.接

著往下看：

//loadtheserverfromtheframeworkclassloader

Thread.currenf777read().setContextClassLoader(frameworkClassLoader);

Class<?>jettyServer=Class.for/VameCorg.apache.ni??.web.server.JettyServer.true,frameworkClassLoader);

Constructor<?>jettyConstructor=jettyServer.getConstructor(NiFiProperties.class.Set.class);

finallongstartTime=System.nanoT/mef);

nifiServer=(NiFiServer)jettyConstructor.newlnstance(properties.narBundles):

nifiServer.setExtensionMapping(extensionMapping);

nifiServer.setBundles(systemBundle.narBundles);

回想架構圖VM的最上層是webserver,這個webserver就是在這里被加載了，這是?個jetty

server，繼續往下看：

if(shutdown){

LOGG￡Rinfo("NiFihasbeenshutdownviaNiFiBootstrap.WillnotstartController");

}else{

nifiServer.start();

if(bootstrapListener!=null){

bootstrapListener.sendStartedStatus(true);

)

finallongduration=System.nanoT/meO?startTime:

LOGGE/7.info("Controllerinitializationtook"+duration+"nanoseconds"

+"("+(int)TimeUnit.SECO/VDS.convert(duration.T\rne\Jn\{.NANOSECONDS)+"seconds)."):

)

start這個nifiServer,這個NiFi對象的構造方法這里就全部走完了.

2.1.4.NiFi-Web

NiFiServer.start。的方法內，代碼跳轉到nifi-framework下的一個了?模塊

nifi-web內了。

〈parent〉

?*2nifi-user-actions

<groupld>org.apache.nifi</groupld>

▼*2nifi-web<artifactld>nifi-framework</artifactld>

?*2nifi-custom-ui-utilities<version>l.10.0</version>

?nifi-jetty</parent>

?nifi-ui-extension<artifactld>nifi-web</artifactld>

?*￡nifi-web-api<packaging>pom</packaging>

?*2nifi-web-content-access<modules>

?nifi-web-content-viewer<module>nifi-web-optimistic-l.ocking</inodule>

<module>nifi-custom-ui-utilities</modul.e>

?*￡nifi?web-docs

<module>nifi-web-security</modul.e>

?*2nifi<web-error

<module>nifi-web-api</module>

?*2nifi-web-optimistic-locking<module>nifi-web-error</module>

?nifi-web-security<module>nifi-web-docs</modul.e>

?*￡nifi-web-ui<module>nifi-web-content-viewer</module>

??target<module>nifi-web-ui</modul.e>

他nifi-web.iml<module>nifi-jetty</module>

mpom.xml<n>odule>nifi-web-content-access</module>

<module>nifi-ui-extension</module>

?由target

〈/modules〉

_____rUCraEawcrUIEI

2.1.5.nifi-jetty

與Web相關的代碼都在這個模塊了，包括Server和界面相關的代碼，上面提到的NiFiServer的實現

類JettyServer就在子模塊nifi-jetty內了。

▼!(nifi-web>1

?*2nifl*custom*ui*utilities

937eOverride

▼hsrc9385publicvoid|rtrt(|>{

try《

▼?main

〃Createastandardextensionmanageranddiscoverextensions

?feidssembly

finalExtensionDiscoveringMandgerextensioritanager=newStandardExtensionOisc

extensiorManager.dIscoverExtension*(systertundle,bundles)：

▼Biorg.apache.nifi.web.serverextenslorHaMQer.logClas$Loaderttapplng();

?GHostHeaderHandler

▼OJettyServerJava//Settheextensionmanagerintotheholderwhichmakesitavailabletothe5

??JettyServerExtensiorHanagerttol.der.jnit(extensiorManager);

??ServerConnectorCreator

?GServerConfigurationExceptionHGeneratedocsforextensions

DoeGenerator.gener?te(props,extensionManager,extensionKapping);

?feSresources

?totest

接著看JettyServer這個類，上面的NiFi構造方法里面最后是先實例化了這個JettyServer,然

后調用了start方法.先看它的構造方法，只看注釋，找到了核心方法：

//loadwarsfromthebundle

finalHandlerwarHandlers=loadlnitialWars(bundles),

可以看到，其實就是把war包加載進來了，這些war包就是nifi-web下面的子工程，有幾個子工

程的pom文件中配置的就是<packaging>war〈/packaging>

接著看這個Start方法：

第一句就是ExtensionManager.discoverExtensions(systemBundle,bundles);就是這里把所有的

擴展類加載進JVM了，看到看到ExtensionManager的注釋，這個注釋就說明了一切

ScansthroughtheclasspathtoloadallFlowFHeProcessors,FlowFileComparators,andReportingTasksusingthe

serviceproviderAPIandrunningthroughallclassloaders(root,NARs).

這個ExtensionManager在加載類的時候,用到了java的一種比較高級的機制，java

SPI(serviceproviderinterface),這種機制在很多框架中比如spring中大量使用

finalServiceLoader<?>serviceLoader=ServiceLoader./oad(entry.getKey().bundle.getClassLoader()):

這個機制解釋了為什么寫自定義的處理器的時候要在/resources/META-INF/services目錄下面

寫上配置文件.在自定義處理開發的時候，一定要注意寫這個配置文件，否則類是加載不進來的

接著start這個jettyserver,接著往下看，只看注釋，可以看到,大致就是做了