數(shù)據(jù)集成工具：Apache Nifi：數(shù)據(jù)流與處理器理解

數(shù)據(jù)集成工具：ApacheNifi：數(shù)據(jù)流與處理器理解1數(shù)據(jù)集成工具：ApacheNifi：數(shù)據(jù)流與處理器理解1.1介紹ApacheNifi1.1.1Nifi的歷史與發(fā)展ApacheNifi是一個易于使用、功能強(qiáng)大且可靠的數(shù)據(jù)處理和分發(fā)系統(tǒng)。它由美國國家安全局（NSA）開發(fā)，并于2014年開源，隨后被Apache軟件基金會接納為頂級項(xiàng)目。Nifi的設(shè)計(jì)初衷是為了自動化數(shù)據(jù)流的處理，提供一種靈活的方式來管理數(shù)據(jù)的收集、聚合和分發(fā)。它支持實(shí)時和批量數(shù)據(jù)處理，能夠處理各種類型和格式的數(shù)據(jù)，包括結(jié)構(gòu)化、半結(jié)構(gòu)化和非結(jié)構(gòu)化數(shù)據(jù)。1.1.2Nifi的核心概念與架構(gòu)核心概念數(shù)據(jù)流：在Nifi中，數(shù)據(jù)流是指數(shù)據(jù)從源頭到目的地的移動路徑。數(shù)據(jù)流由處理器、連接、控制器服務(wù)和策略組成。處理器：處理器是Nifi中的核心組件，負(fù)責(zé)執(zhí)行數(shù)據(jù)流中的操作，如讀取、寫入、轉(zhuǎn)換、路由等。每個處理器都有特定的功能，可以被配置和連接以創(chuàng)建復(fù)雜的數(shù)據(jù)處理流程。連接：連接是處理器之間的數(shù)據(jù)傳輸通道。數(shù)據(jù)通過連接從一個處理器流向另一個處理器?？刂破鞣?wù)：控制器服務(wù)提供配置信息，如數(shù)據(jù)庫連接、加密服務(wù)等，用于支持處理器的運(yùn)行。策略：策略包括數(shù)據(jù)流的控制策略和數(shù)據(jù)的訪問控制策略，確保數(shù)據(jù)流的正確執(zhí)行和數(shù)據(jù)的安全。架構(gòu)Nifi的架構(gòu)設(shè)計(jì)為分布式、可擴(kuò)展和容錯的。它包括以下關(guān)鍵組件：NiFi集群：Nifi可以部署在單個節(jié)點(diǎn)上，也可以部署在集群中以提高性能和可靠性。集群中的節(jié)點(diǎn)通過共享數(shù)據(jù)流和狀態(tài)信息來協(xié)同工作。NiFi節(jié)點(diǎn)：每個Nifi節(jié)點(diǎn)都是一個獨(dú)立的運(yùn)行實(shí)例，可以處理數(shù)據(jù)流的一部分。NiFiUI：Nifi提供了一個直觀的用戶界面，用于設(shè)計(jì)、監(jiān)控和管理數(shù)據(jù)流。用戶可以通過拖放操作來創(chuàng)建和配置處理器，以及查看數(shù)據(jù)流的實(shí)時狀態(tài)。NiFiRESTAPI：除了UI，Nifi還提供了一個RESTAPI，允許用戶通過編程方式來管理數(shù)據(jù)流和處理器。這為自動化和集成提供了便利。示例：使用Nifi處理器處理數(shù)據(jù)假設(shè)我們有一個數(shù)據(jù)流，需要從一個CSV文件中讀取數(shù)據(jù)，然后將數(shù)據(jù)轉(zhuǎn)換為JSON格式，并最終將數(shù)據(jù)寫入到一個數(shù)據(jù)庫中。以下是使用Nifi處理器實(shí)現(xiàn)這一數(shù)據(jù)流的步驟：GetFile處理器：配置GetFile處理器來讀取CSV文件。設(shè)置輸入目錄為CSV文件所在的目錄，輸出目錄為Nifi內(nèi)部的臨時目錄。ConvertRecord處理器：使用ConvertRecord處理器將CSV格式的數(shù)據(jù)轉(zhuǎn)換為JSON格式。配置轉(zhuǎn)換規(guī)則，指定CSV到JSON的轉(zhuǎn)換邏輯。PutDatabaseRecord處理器：配置PutDatabaseRecord處理器將JSON數(shù)據(jù)寫入到數(shù)據(jù)庫中。設(shè)置數(shù)據(jù)庫連接信息，包括數(shù)據(jù)庫類型、連接字符串、用戶名和密碼。代碼示例：使用NifiRESTAPI創(chuàng)建處理器curl-XPOST\

'http://localhost:8080/nifi-api/process-groups/root/processors'\

-H'Content-Type:application/json'\

-d'{

"component":{

"name":"GetFile",

"type":"cessors.standard.GetFile",

"bundle":{

"groupId":"org.apache.nifi",

"artifactId":"nifi-standard-nar",

"version":"1.13.0"

},

"position":{

"x":100.0,

"y":200.0

},

"configurableProperties":{

"inputDirectory":"/path/to/csv/files",

"outputDirectory":"/path/to/nifi/temp"

}

}

}'這段代碼使用curl命令通過Nifi的RESTAPI創(chuàng)建了一個GetFile處理器。處理器被命名為“GetFile”，并被配置為從指定的目錄讀取CSV文件，并將數(shù)據(jù)輸出到Nifi的臨時目錄。通過理解Nifi的歷史、核心概念和架構(gòu)，以及實(shí)際操作中的示例，我們可以更好地利用Nifi來構(gòu)建和管理復(fù)雜的數(shù)據(jù)集成流程。Nifi的靈活性和可擴(kuò)展性使其成為處理大數(shù)據(jù)和實(shí)時數(shù)據(jù)流的理想工具。2理解數(shù)據(jù)流2.1數(shù)據(jù)流的基本原理數(shù)據(jù)流（DataFlow）是數(shù)據(jù)集成工具中的核心概念，它描述了數(shù)據(jù)如何在系統(tǒng)中從一個點(diǎn)移動到另一個點(diǎn)的過程。在ApacheNiFi中，數(shù)據(jù)流是由一系列的處理器（Processor）、連接（Connection）、輸入/輸出端口（Input/OutputPort）以及工作流（Workflow）組成的，它們共同協(xié)作來實(shí)現(xiàn)數(shù)據(jù)的采集、處理和分發(fā)。2.1.1處理器（Processor）處理器是數(shù)據(jù)流中的基本執(zhí)行單元，負(fù)責(zé)執(zhí)行特定的數(shù)據(jù)處理任務(wù)。例如，GetFile處理器用于從文件系統(tǒng)中讀取數(shù)據(jù)，PutSQL處理器用于將數(shù)據(jù)寫入數(shù)據(jù)庫。每個處理器都有其特定的配置屬性，這些屬性決定了處理器如何執(zhí)行其任務(wù)。2.1.2連接（Connection）連接是處理器之間的數(shù)據(jù)傳輸通道。數(shù)據(jù)從一個處理器輸出后，通過連接傳遞給下一個處理器。連接可以配置為只傳遞滿足特定條件的數(shù)據(jù)，例如，使用success、failure或自定義的關(guān)系來決定數(shù)據(jù)的流向。2.1.3輸入/輸出端口（Input/OutputPort）端口用于在NiFi實(shí)例之間傳輸數(shù)據(jù)。輸入端口接收來自外部的數(shù)據(jù)，而輸出端口將數(shù)據(jù)發(fā)送到外部系統(tǒng)。端口可以配置為安全的，以確保數(shù)據(jù)傳輸?shù)陌踩浴?.1.4工作流（Workflow）工作流是處理器、連接和端口的組合，它定義了數(shù)據(jù)流的邏輯路徑。通過設(shè)計(jì)不同的工作流，可以實(shí)現(xiàn)復(fù)雜的數(shù)據(jù)處理和分發(fā)策略。2.2創(chuàng)建和配置數(shù)據(jù)流在ApacheNiFi中創(chuàng)建和配置數(shù)據(jù)流涉及以下步驟：啟動NiFi并訪問NiFi界面：首先，確保ApacheNiFi服務(wù)正在運(yùn)行，并通過瀏覽器訪問NiFi的WebUI。創(chuàng)建處理器：在NiFi的畫布上，通過右鍵點(diǎn)擊并選擇“創(chuàng)建處理器”來添加處理器。例如，添加GetFile處理器來讀取文件系統(tǒng)中的數(shù)據(jù)。配置處理器：雙擊處理器圖標(biāo)，打開配置窗口。在這里，可以設(shè)置處理器的屬性，如GetFile的輸入目錄屬性，用于指定要讀取文件的目錄。創(chuàng)建連接：通過拖拽從一個處理器到另一個處理器的箭頭來創(chuàng)建連接。例如，從GetFile處理器拖拽到PutSQL處理器，以將讀取的文件數(shù)據(jù)寫入數(shù)據(jù)庫。配置連接：在連接上右鍵點(diǎn)擊，選擇“配置”，可以設(shè)置連接的屬性，如數(shù)據(jù)傳遞的條件和優(yōu)先級。添加輸入/輸出端口：如果需要在NiFi實(shí)例之間傳輸數(shù)據(jù)，可以添加輸入和輸出端口。例如，添加InputPort來接收來自外部的數(shù)據(jù)流。配置端口：端口的配置包括設(shè)置其監(jiān)聽的地址和端口，以及安全設(shè)置，如SSL證書的使用。啟動數(shù)據(jù)流：配置完成后，確保所有處理器都處于“運(yùn)行”狀態(tài)，然后啟動數(shù)據(jù)流，觀察數(shù)據(jù)如何在系統(tǒng)中流動。2.2.1示例：使用GetFile和PutSQL處理器創(chuàng)建數(shù)據(jù)流####步驟1：添加`GetFile`處理器

-在NiFi畫布上添加`GetFile`處理器。

-配置`GetFile`處理器的`輸入目錄`屬性，指向包含數(shù)據(jù)文件的目錄。

####步驟2：添加`PutSQL`處理器

-在NiFi畫布上添加`PutSQL`處理器。

-配置`PutSQL`處理器的數(shù)據(jù)庫連接信息，包括數(shù)據(jù)庫類型、URL、用戶名和密碼。

####步驟3：創(chuàng)建連接

-從`GetFile`處理器拖拽到`PutSQL`處理器，創(chuàng)建連接。

####步驟4：配置連接

-確保連接的傳遞關(guān)系設(shè)置為`success`，以便`GetFile`處理器成功讀取的數(shù)據(jù)能夠傳遞給`PutSQL`處理器。

####步驟5：啟動數(shù)據(jù)流

-將所有處理器設(shè)置為“運(yùn)行”狀態(tài)，啟動數(shù)據(jù)流。通過以上步驟，可以創(chuàng)建一個從文件系統(tǒng)讀取數(shù)據(jù)，并將其寫入數(shù)據(jù)庫的數(shù)據(jù)流。在實(shí)際操作中，還需要根據(jù)具體的數(shù)據(jù)格式和數(shù)據(jù)庫結(jié)構(gòu)來調(diào)整處理器的配置，以確保數(shù)據(jù)能夠正確地被讀取和寫入。2.2.2注意事項(xiàng)在設(shè)計(jì)數(shù)據(jù)流時，應(yīng)考慮數(shù)據(jù)的處理順序和依賴關(guān)系，確保數(shù)據(jù)流的邏輯正確。配置處理器和連接時，應(yīng)仔細(xì)檢查屬性設(shè)置，避免數(shù)據(jù)處理錯誤或數(shù)據(jù)丟失。定期監(jiān)控?cái)?shù)據(jù)流的運(yùn)行狀態(tài)，及時發(fā)現(xiàn)并解決問題，確保數(shù)據(jù)處理的連續(xù)性和穩(wěn)定性。通過理解和掌握ApacheNiFi中的數(shù)據(jù)流和處理器，可以有效地設(shè)計(jì)和實(shí)現(xiàn)數(shù)據(jù)集成解決方案，滿足各種數(shù)據(jù)處理和分發(fā)的需求。3數(shù)據(jù)集成工具：ApacheNifi：深入處理器3.1處理器的功能與分類在ApacheNifi中，處理器是數(shù)據(jù)流的核心組件，負(fù)責(zé)執(zhí)行數(shù)據(jù)處理任務(wù)。它們可以被配置為執(zhí)行各種操作，如讀取、寫入、轉(zhuǎn)換、路由、過濾數(shù)據(jù)等。處理器的靈活性和可配置性使得Nifi能夠適應(yīng)廣泛的數(shù)據(jù)集成需求。3.1.1功能讀取數(shù)據(jù)：如GetFile處理器，用于從文件系統(tǒng)中讀取數(shù)據(jù)。寫入數(shù)據(jù)：如PutFile處理器，用于將數(shù)據(jù)寫入文件系統(tǒng)。轉(zhuǎn)換數(shù)據(jù)：如ConvertAttribute處理器，用于修改數(shù)據(jù)流中的屬性。路由數(shù)據(jù)：如RouteOnAttribute處理器，根據(jù)數(shù)據(jù)屬性將數(shù)據(jù)流路由到不同的下游處理器。過濾數(shù)據(jù)：如RemoveAttribute處理器，用于從數(shù)據(jù)流中移除特定屬性。聚合數(shù)據(jù)：如Aggregate處理器，用于合并多個數(shù)據(jù)流到一個流中。3.1.2分類處理器根據(jù)其功能可以分為以下幾類：輸入處理器：用于接收數(shù)據(jù)。輸出處理器：用于發(fā)送數(shù)據(jù)到外部系統(tǒng)。轉(zhuǎn)換處理器：用于修改數(shù)據(jù)內(nèi)容或?qū)傩?。路由處理器：用于根?jù)條件將數(shù)據(jù)流路由到不同的路徑?？刂铺幚砥鳎河糜诠芾鞱ifi的運(yùn)行狀態(tài)或配置。過濾處理器：用于篩選或移除數(shù)據(jù)流中的內(nèi)容或?qū)傩浴?.2常用處理器的使用詳解3.2.1GetFile功能描述GetFile處理器用于從文件系統(tǒng)中讀取文件，并將文件內(nèi)容封裝成FlowFile，以便后續(xù)處理。配置參數(shù)輸入目錄：指定處理器監(jiān)控的目錄，當(dāng)目錄中有新文件時，處理器會自動讀取。輸出目錄：可選，用于指定讀取文件后，文件的移動或刪除位置。文件過濾器：用于指定處理器讀取的文件類型或名稱模式。示例假設(shè)我們有一個GetFile處理器，配置如下：-輸入目錄：/data/input-輸出目錄：/data/processed-文件過濾器：*.txt當(dāng)有新的.txt文件放入/data/input目錄時，GetFile處理器會讀取文件內(nèi)容，創(chuàng)建一個FlowFile，并將其移動到/data/processed目錄。3.2.2PutFile功能描述PutFile處理器用于將FlowFile的內(nèi)容寫入到文件系統(tǒng)中。配置參數(shù)輸出目錄：指定處理器將數(shù)據(jù)寫入的目錄。文件名生成器：用于生成輸出文件的名稱，可以基于FlowFile的屬性或內(nèi)容。示例假設(shè)我們有一個PutFile處理器，配置如下：-輸出目錄：/data/output-文件名生成器：使用SimpleFileNameGenerator，將輸出文件名設(shè)置為output-${uuid()}.txt當(dāng)有FlowFile到達(dá)PutFile處理器時，處理器會將FlowFile的內(nèi)容寫入到/data/output目錄下的一個新文件中，文件名格式為output-UUID.txt。3.2.3ConvertAttribute功能描述ConvertAttribute處理器用于將數(shù)據(jù)流中的屬性從一種類型轉(zhuǎn)換為另一種類型，例如從字符串轉(zhuǎn)換為日期。配置參數(shù)屬性鍵：指定要轉(zhuǎn)換的屬性名稱。轉(zhuǎn)換類型：指定轉(zhuǎn)換的目標(biāo)類型，如String、Date、Integer等。示例假設(shè)我們有一個ConvertAttribute處理器，配置如下：-屬性鍵：timestamp-轉(zhuǎn)換類型：Date如果FlowFile中有一個名為timestamp的屬性，其值為字符串2023-03-15T12:00:00Z，ConvertAttribute處理器會將其轉(zhuǎn)換為日期類型。3.2.4RouteOnAttribute功能描述RouteOnAttribute處理器根據(jù)FlowFile的屬性值，將數(shù)據(jù)流路由到不同的下游處理器。配置參數(shù)屬性鍵：用于路由的屬性名稱。路由策略：定義屬性值與下游處理器之間的映射關(guān)系。示例假設(shè)我們有一個RouteOnAttribute處理器，配置如下：-屬性鍵：priority-路由策略：如果priority屬性值為high，則路由到HighPriorityQueue處理器；如果為low，則路由到LowPriorityQueue處理器。當(dāng)一個FlowFile到達(dá)RouteOnAttribute處理器時，如果其priority屬性值為high，則會被路由到HighPriorityQueue處理器；如果為low，則會被路由到LowPriorityQueue處理器。3.2.5RemoveAttribute功能描述RemoveAttribute處理器用于從FlowFile中移除指定的屬性。配置參數(shù)屬性鍵：指定要移除的屬性名稱。示例假設(shè)我們有一個RemoveAttribute處理器，配置如下：-屬性鍵：unwantedAttribute當(dāng)一個FlowFile到達(dá)RemoveAttribute處理器時，如果其包含unwantedAttribute屬性，該屬性將被移除。3.2.6Aggregate功能描述Aggregate處理器用于合并多個FlowFile到一個FlowFile中，通常用于數(shù)據(jù)聚合或批處理。配置參數(shù)聚合策略：定義如何選擇要聚合的FlowFile。聚合觸發(fā)器：定義何時觸發(fā)聚合操作。示例假設(shè)我們有一個Aggregate處理器，配置如下：-聚合策略：使用TimeTriggeredAggregator，根據(jù)時間窗口聚合FlowFile。-聚合觸發(fā)器：設(shè)置時間窗口為5minutes。當(dāng)有多個FlowFile在5分鐘內(nèi)到達(dá)Aggregate處理器時，它們會被合并成一個FlowFile，然后發(fā)送到下游處理器進(jìn)行進(jìn)一步處理。通過上述處理器的使用，ApacheNifi能夠構(gòu)建復(fù)雜的數(shù)據(jù)流，實(shí)現(xiàn)數(shù)據(jù)的高效集成和處理。每個處理器的配置和使用都需要根據(jù)具體的數(shù)據(jù)處理需求進(jìn)行調(diào)整，以達(dá)到最佳的處理效果。4數(shù)據(jù)流的控制與管理4.1流程控制器的設(shè)置在ApacheNiFi中，流程控制器是數(shù)據(jù)流的核心組件，負(fù)責(zé)管理NiFi實(shí)例的啟動、停止、調(diào)度和配置。它提供了對NiFi流程的全局控制，確保數(shù)據(jù)流的穩(wěn)定性和可靠性。流程控制器的設(shè)置主要包括以下幾個方面：調(diào)度策略：決定NiFi處理器的執(zhí)行頻率和方式。例如，可以設(shè)置處理器每5分鐘執(zhí)行一次，或者在接收到數(shù)據(jù)時立即執(zhí)行。線程數(shù)量：控制同時執(zhí)行的處理器線程數(shù)，影響數(shù)據(jù)流的并行處理能力。數(shù)據(jù)流優(yōu)先級：通過設(shè)置優(yōu)先級，可以控制不同數(shù)據(jù)流的執(zhí)行順序，確保關(guān)鍵數(shù)據(jù)流的優(yōu)先執(zhí)行。故障恢復(fù)：配置數(shù)據(jù)流在遇到故障時的恢復(fù)策略，如自動重啟、數(shù)據(jù)重試等。4.1.1示例：設(shè)置處理器的調(diào)度策略在NiFi的流程編輯器中，選擇一個處理器，然后在右側(cè)的屬性面板中找到“調(diào)度”部分。這里可以設(shè)置處理器的調(diào)度策略，例如：調(diào)度策略：事件驅(qū)動

調(diào)度時間間隔：5分鐘

運(yùn)行時間：立即這意味著處理器將在接收到數(shù)據(jù)時立即執(zhí)行，并且如果沒有數(shù)據(jù)接收，它將每5分鐘檢查一次并執(zhí)行。4.2數(shù)據(jù)流的監(jiān)控與調(diào)試數(shù)據(jù)流的監(jiān)控與調(diào)試是確保數(shù)據(jù)集成流程正確性和效率的關(guān)鍵。NiFi提供了豐富的工具和界面，幫助用戶監(jiān)控?cái)?shù)據(jù)流的狀態(tài)，以及在出現(xiàn)問題時進(jìn)行調(diào)試。監(jiān)控工具：NiFi的監(jiān)控工具可以顯示數(shù)據(jù)流的實(shí)時狀態(tài)，包括處理器的執(zhí)行情況、數(shù)據(jù)傳輸速率、系統(tǒng)資源使用情況等。日志記錄：NiFi記錄詳細(xì)的日志，包括數(shù)據(jù)流的運(yùn)行日志、系統(tǒng)日志和審計(jì)日志，幫助用戶追蹤數(shù)據(jù)流的執(zhí)行歷史和系統(tǒng)行為。調(diào)試模式：通過啟用調(diào)試模式，用戶可以查看數(shù)據(jù)流中每個組件的詳細(xì)狀態(tài)，包括輸入和輸出數(shù)據(jù)的內(nèi)容，這對于調(diào)試復(fù)雜的數(shù)據(jù)流非常有幫助。4.2.1示例：使用NiFi的監(jiān)控工具在NiFi的Web界面中，可以直觀地看到數(shù)據(jù)流的運(yùn)行狀態(tài)。例如，點(diǎn)擊一個處理器，可以看到其“狀態(tài)”面板，顯示了處理器的執(zhí)行次數(shù)、成功次數(shù)、失敗次數(shù)等信息。此外，還可以查看“數(shù)據(jù)包隊(duì)列”和“數(shù)據(jù)包狀態(tài)”，了解數(shù)據(jù)包在處理器中的處理情況。監(jiān)控面板顯示：

-處理器執(zhí)行次數(shù)：1000

-成功次數(shù)：990

-失敗次數(shù)：10這表明處理器已經(jīng)執(zhí)行了1000次，其中990次成功，10次失敗，可能需要進(jìn)一步檢查失敗的原因。4.2.2示例：啟用NiFi的調(diào)試模式在NiFi中，可以通過設(shè)置處理器的“調(diào)試”屬性來啟用調(diào)試模式。例如，設(shè)置“調(diào)試記錄級別”為“詳細(xì)”，并選擇“調(diào)試記錄條件”為“總是記錄”，這樣處理器在執(zhí)行時會記錄詳細(xì)的調(diào)試信息。調(diào)試記錄級別：詳細(xì)

調(diào)試記錄條件：總是記錄啟用調(diào)試模式后，可以在“日志”面板中查看調(diào)試信息，幫助定位和解決問題。通過上述設(shè)置和工具，用戶可以有效地控制和管理ApacheNiFi中的數(shù)據(jù)流，確保數(shù)據(jù)集成任務(wù)的順利進(jìn)行。監(jiān)控和調(diào)試功能的使用，可以提高數(shù)據(jù)流的可見性和可控性，減少故障發(fā)生，提升數(shù)據(jù)處理的效率和質(zhì)量。5數(shù)據(jù)集成實(shí)踐5.1數(shù)據(jù)源與目標(biāo)的連接在數(shù)據(jù)集成項(xiàng)目中，ApacheNiFi是一個強(qiáng)大的工具，用于管理數(shù)據(jù)流。數(shù)據(jù)源與目標(biāo)的連接是數(shù)據(jù)集成流程中的關(guān)鍵步驟，涉及到數(shù)據(jù)的獲取和傳輸至目標(biāo)系統(tǒng)。NiFi通過其豐富的處理器集合，提供了靈活的方式來連接各種數(shù)據(jù)源和目標(biāo)。5.1.1數(shù)據(jù)源處理器GetFile功能：從文件系統(tǒng)中讀取文件。配置：指定文件目錄、文件過濾器等。示例：假設(shè)我們需要從一個目錄中讀取CSV文件。<!--GetFile處理器配置-->

<processorclass="cessors.standard.GetFile">

<name>GetCSVFiles</name>

<inputPorts/>

<outputPorts/>

<scheduling>

<schedulingPeriod>1sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<FileFilter>.*\.csv</FileFilter>

<InputDirectory>/path/to/csv/files</InputDirectory>

<KeepSourceFile>False</KeepSourceFile>

<MaxFilestoRead>1</MaxFilestoRead>

<MaxFileAge>0sec</MaxFileAge>

<MaxFileSize>10MB</MaxFileSize>

<MaxQueueSize>1000</MaxQueueSize>

<PollingInterval>1sec</PollingInterval>

<ReadBufferSize>1MB</ReadBufferSize>

<ReadLineCount>10000</ReadLineCount>

<ReadLineTimeout>10sec</ReadLineTimeout>

<ReadLineTimeoutYield>1sec</ReadLineTimeoutYield>

<ReadLineTimeoutYieldPeriod>1sec</ReadLineTimeoutYieldPeriod>

<ReadLineTimeoutYieldStrategy>Yield</ReadLineTimeoutYieldStrategy>

<ReadLineTimeoutYieldTime>1sec</ReadLineTimeoutYieldTime>

<ReadLineTimeoutYieldTimeUnit>SECONDS</ReadLineTimeoutYieldTimeUnit>

<ReadLineTimeoutYielding>False</ReadLineTimeoutYielding>

<ReadLineTimeoutYieldingStrategy>Yield</ReadLineTimeoutYieldingStrategy>

<ReadLineTimeoutYieldingTime>1sec</ReadLineTimeoutYieldingTime>

<ReadLineTimeoutYieldingTimeUnit>SECONDS</ReadLineTimeoutYieldingTimeUnit>

<ReadLineTimeoutYieldingYield>1sec</ReadLineTimeoutYieldingYield>

<ReadLineTimeoutYieldingYieldPeriod>1sec</ReadLineTimeoutYieldingYieldPeriod>

<ReadLineTimeoutYieldingYieldStrategy>Yield</ReadLineTimeoutYieldingYieldStrategy>

<ReadLineTimeoutYieldingYieldTime>1sec</ReadLineTimeoutYieldingYieldTime>

<ReadLineTimeoutYieldingYieldTimeUnit>SECONDS</ReadLineTimeoutYieldingYieldTimeUnit>

<ReadLineTimeoutYieldingYielding>False</ReadLineTimeoutYieldingYielding>

<ReadLineTimeoutYieldingYieldingStrategy>Yield</ReadLineTimeoutYieldingYieldingStrategy>

<ReadLineTimeoutYieldingYieldingTime>1sec</ReadLineTimeoutYieldingYieldingTime>

<ReadLineTimeoutYieldingYieldingTimeUnit>SECONDS</ReadLineTimeoutYieldingYieldingTimeUnit>

<ReadLineTimeoutYieldingYieldingYield>1sec</ReadLineTimeoutYieldingYieldingYield>

<ReadLineTimeoutYieldingYieldingYieldPeriod>1sec</ReadLineTimeoutYieldingYieldingYieldPeriod>

<ReadLineTimeoutYieldingYieldingYieldStrategy>Yield</ReadLineTimeoutYieldingYieldingYieldStrategy>

<ReadLineTimeoutYieldingYieldingYieldTime>1sec</ReadLineTimeoutYieldingYieldingYieldTime>

<ReadLineTimeoutYieldingYieldingYieldTimeUnit>SECONDS</ReadLineTimeoutYieldingYieldingYieldTimeUnit>

</properties>

</processor>JDBCDatabaseTableReader功能：從數(shù)據(jù)庫讀取數(shù)據(jù)。配置：數(shù)據(jù)庫連接信息、查詢語句等。示例：從MySQL數(shù)據(jù)庫讀取數(shù)據(jù)。<!--JDBCDatabaseTableReader處理器配置-->

<processorclass="cessors.jdbc.JDBCDatabaseTableReader">

<name>ReadfromMySQL</name>

<inputPorts/>

<outputPorts/>

<scheduling>

<schedulingPeriod>1min</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<ConnectionPoolingService>MyDatabaseConnectionPool</ConnectionPoolingService>

<FetchSize>100</FetchSize>

<Query>SELECT*FROMmy_table</Query>

<ResultSetRowReader>StandardResultSetRowReader</ResultSetRowReader>

<RowReaderProperties>MyRowReaderProperties</RowReaderProperties>

<RowReaderService>MyRowReaderService</RowReaderService>

<StatementCacheService>MyStatementCacheService</StatementCacheService>

<TransactionSize>100</TransactionSize>

</properties>

</processor>5.1.2數(shù)據(jù)目標(biāo)處理器PutFile功能：將數(shù)據(jù)寫入文件系統(tǒng)。配置：目標(biāo)目錄、文件名生成策略等。示例：將處理后的數(shù)據(jù)寫入另一個目錄。<!--PutFile處理器配置-->

<processorclass="cessors.standard.PutFile">

<name>WritetoOutputDirectory</name>

<inputPorts/>

<outputPorts/>

<scheduling>

<schedulingPeriod>1sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<FileFilter>.*</FileFilter>

<FileWriter>StandardFileWriter</FileWriter>

<FileWriterProperties>MyFileWriterProperties</FileWriterProperties>

<FileWriterService>MyFileWriterService</FileWriterService>

<OutputDirectory>/path/to/output/directory</OutputDirectory>

<FileNameGenerator>StandardFileNameGenerator</FileNameGenerator>

<FileNameGeneratorProperties>MyFileNameGeneratorProperties</FileNameGeneratorProperties>

<FileNameGeneratorService>MyFileNameGeneratorService</FileNameGeneratorService>

<MaxFileAge>0sec</MaxFileAge>

<MaxFileSize>10MB</MaxFileSize>

<MaxQueueSize>1000</MaxQueueSize>

<PollingInterval>1sec</PollingInterval>

<ReadBufferSize>1MB</ReadBufferSize>

<ReadLineCount>10000</ReadLineCount>

<ReadLineTimeout>10sec</ReadLineTimeout>

<ReadLineTimeoutYield>1sec</ReadLineTimeoutYield>

<ReadLineTimeoutYieldPeriod>1sec</ReadLineTimeoutYieldPeriod>

<ReadLineTimeoutYieldStrategy>Yield</ReadLineTimeoutYieldStrategy>

<ReadLineTimeoutYieldTime>1sec</ReadLineTimeoutYieldTime>

<ReadLineTimeoutYieldTimeUnit>SECONDS</ReadLineTimeoutYieldTimeUnit>

<ReadLineTimeoutYielding>False</ReadLineTimeoutYielding>

<ReadLineTimeoutYieldingStrategy>Yield</ReadLineTimeoutYieldingStrategy>

<ReadLineTimeoutYieldingTime>1sec</ReadLineTimeoutYieldingTime>

<ReadLineTimeoutYieldingTimeUnit>SECONDS</ReadLineTimeoutYieldingTimeUnit>

<ReadLineTimeoutYieldingYield>1sec</ReadLineTimeoutYieldingYield>

<ReadLineTimeoutYieldingYieldPeriod>1sec</ReadLineTimeoutYieldingYieldPeriod>

<ReadLineTimeoutYieldingYieldStrategy>Yield</ReadLineTimeoutYieldingYieldStrategy>

<ReadLineTimeoutYieldingYieldTime>1sec</ReadLineTimeoutYieldingYieldTime>

<ReadLineTimeoutYieldingYieldTimeUnit>SECONDS</ReadLineTimeoutYieldingYieldTimeUnit>

</properties>

</processor>JDBCDatabaseTableWriter功能：將數(shù)據(jù)寫入數(shù)據(jù)庫。配置：數(shù)據(jù)庫連接信息、寫入表名等。示例：將數(shù)據(jù)寫入PostgreSQL數(shù)據(jù)庫。<!--JDBCDatabaseTableWriter處理器配置-->

<processorclass="cessors.jdbc.JDBCDatabaseTableWriter">

<name>WritetoPostgreSQL</name>

<inputPorts/>

<outputPorts/>

<scheduling>

<schedulingPeriod>1sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<ConnectionPoolingService>MyDatabaseConnectionPool</ConnectionPoolingService>

<TableWriter>StandardTableWriter</TableWriter>

<TableWriterProperties>MyTableWriterProperties</TableWriterProperties>

<TableWriterService>MyTableWriterService</TableWriterService>

<TargetTable>my_target_table</TargetTable>

<WriteStrategy>Batch</WriteStrategy>

<BatchSize>100</BatchSize>

<BatchTimeout>1min</BatchTimeout>

</properties>

</processor>5.2數(shù)據(jù)轉(zhuǎn)換與路由策略數(shù)據(jù)轉(zhuǎn)換與路由是數(shù)據(jù)集成中的核心環(huán)節(jié)，ApacheNiFi提供了多種處理器來實(shí)現(xiàn)數(shù)據(jù)的轉(zhuǎn)換和基于條件的路由。5.2.1數(shù)據(jù)轉(zhuǎn)換處理器UpdateAttribute功能：更新或添加流文件的屬性。配置：屬性更新規(guī)則。示例：更新文件名屬性。<!--UpdateAttribute處理器配置-->

<processorclass="cessors.standard.UpdateAttribute">

<name>UpdateFileName</name>

<inputPorts/>

<outputPorts/>

<scheduling>

<schedulingPeriod>1sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<AttributeExpressionLanguage>filename+"_processed"</AttributeExpressionLanguage>

<AttributetoUpdate>filename</AttributetoUpdate>

</properties>

</processor>ExecuteGroovyScript功能：使用Groovy腳本進(jìn)行復(fù)雜的數(shù)據(jù)轉(zhuǎn)換。配置：Groovy腳本、腳本引擎等。示例：使用Groovy腳本將JSON數(shù)據(jù)轉(zhuǎn)換為XML。<!--ExecuteGroovyScript處理器配置-->

<processorclass="cessors.script.ExecuteGroovyScript">

<name>JSONtoXML</name>

<inputPorts/>

<outputPorts/>

<scheduling>

<schedulingPeriod>1sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<ScriptEngine>groovy</ScriptEngine>

<ScriptBody>

importgroovy.json.JsonSlurper

importgroovy.xml.MarkupBuilder

defslurper=newJsonSlurper()

defjson=slurper.parseText(flowFile.content)

defxml=newMarkupBuilder().bind{

root{

json.each{k,v->

node(k){v}

}

}

}

flowFile.content=xml.toString()

</ScriptBody>

</properties>

</processor>5.2.2數(shù)據(jù)路由策略RouteOnAttribute功能：基于流文件屬性的條件路由。配置：屬性名稱、路由條件等。示例：根據(jù)文件類型屬性路由至不同處理器。<!--RouteOnAttribute處理器配置-->

<processorclass="cessors.standard.RouteOnAttribute">

<name>RoutebyFileType</name>

<inputPorts/>

<outputPorts>

<outputPortid="00000000-0000-0000-0000-000000000001">

<name>CSV</name>

<schedulingStrategy>EVENT_DRIVEN</schedulingStrategy>

<schedulingPeriod>0sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

<properties>

<AttributeExpressionLanguage>mime.type=="text/csv"</AttributeExpressionLanguage>

</properties>

</outputPort>

<outputPortid="00000000-0000-0000-0000-000000000002">

<name>JSON</name>

<schedulingStrategy>EVENT_DRIVEN</schedulingStrategy>

<schedulingPeriod>0sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

<properties>

<AttributeExpressionLanguage>mime.type=="application/json"</AttributeExpressionLanguage>

</properties>

</outputPort>

</outputPorts>

<scheduling>

<schedulingPeriod>1sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<RoutingStrategy>Standard</RoutingStrategy>

</properties>

</processor>SplitText功能：根據(jù)指定規(guī)則將文本數(shù)據(jù)分割成多個流文件。配置：分割規(guī)則、分割數(shù)量等。示例：將大文件分割成多個小文件。<!--SplitText處理器配置-->

<processorclass="cessors.standard.SplitText">

<name>SplitLargeFile</name>

<inputPorts/>

<outputPorts/>

<scheduling>

<schedulingPeriod>1sec</schedulingPeriod>

<penalizationPeriod>30sec</penalizationPeriod>

<yieldPeriod>1sec</yieldPeriod>

</scheduling>

<properties>

<BulletinThreshold>0sec</BulletinThreshold>

<LineSplitCount>1000</LineSplitCount>

<MaxChunkSize>1MB</MaxChunkSize>

<MaxChunkAge>10min</MaxChunkAge>

<MaxChunkCount>100</MaxChunkCount>

<MaxQueueSize>1000</MaxQueueSize>

<ReadBufferSize>1MB</ReadBufferSize>

<ReadLineCount>10000</ReadLineCount>

<ReadLineTimeout>10sec</ReadLineTimeout>

<ReadLineTimeoutYield>1sec</ReadLineTimeoutYield>

<ReadLineTimeoutYieldPeriod>1sec</ReadLineTimeoutYieldPeriod>

<ReadLineTimeoutYieldStrategy>Yield</ReadLineTimeoutYieldStrategy>

<ReadLineTimeoutYieldTime>1sec</ReadLineTimeoutYieldTime>

<ReadLineTimeoutYieldTimeUnit>SECONDS</ReadLineTimeoutYieldTimeUnit>

<ReadLineTimeoutYielding>False</ReadLineTimeoutYielding>

<ReadLineTimeoutYieldingStrategy>Yield</ReadLineTimeoutYieldingStrategy>

<ReadLineTimeoutYieldingTime>1sec</ReadLineTimeoutYieldingTime>

<ReadLineTimeoutYieldingTimeUnit>SECONDS</ReadLineTimeoutYieldingTimeUnit>

<ReadLineTimeoutYieldingYield>1sec</ReadLineTimeoutYieldingYield>

<ReadLineTimeoutYieldingYieldPeriod>1sec</ReadLineTimeoutYieldingYieldPeriod>

<ReadLineTimeoutYieldingYieldStrategy>Yield</ReadLineTimeoutYieldingYieldStrategy>

<ReadLineTimeoutYieldingYieldTime>1sec</ReadLineTimeoutYieldingYieldTime>

<ReadLineTimeoutYieldingYieldTimeUnit>SECONDS</ReadLineTimeoutYieldingYieldTimeUnit>

</properties>

</processor>通過上述示例，我們可以看到ApacheNiFi如何通過不同的處理器實(shí)現(xiàn)數(shù)據(jù)源與目標(biāo)的連接，以及數(shù)據(jù)的轉(zhuǎn)換和路由。這些處理器的靈活配置和組合，使得NiFi能夠適應(yīng)各種復(fù)雜的數(shù)據(jù)集成場景。6高級數(shù)據(jù)流設(shè)計(jì)6.1數(shù)據(jù)流的優(yōu)化技巧在ApacheNiFi中，數(shù)據(jù)流的優(yōu)化是確保數(shù)據(jù)處理高效、可靠的關(guān)鍵。以下是一些優(yōu)化技巧：6.1.1使用適當(dāng)?shù)奶幚砥鬟x擇合適的處理器數(shù)據(jù)獲?。菏褂肎etFile或GetKafka等處理器，根據(jù)數(shù)據(jù)源選擇。數(shù)據(jù)轉(zhuǎn)換：如PutKafkaTopic或InvokeHTTP，確保數(shù)據(jù)格式符合下游系統(tǒng)需求。數(shù)據(jù)存儲：PutFile或PutS3Object等，根據(jù)目標(biāo)存儲選擇。示例：使用GetFile處理器<processorid="12345678-90ab-cdef-1234-567890abcdef">

<type>cessors.standard.GetFile</type>

<name>GetFileExample</name>

<properties>

<InputDirectory>/path/to/input/directory</InputDirectory>

<FileFilter>.*\.csv</FileFilter>

</properties>

</processor>此配置示例展示了如何使用GetFile處理器從指定目錄讀取CSV文件。6.1.2調(diào)整線程數(shù)線程池配置線程數(shù)：增加線程數(shù)可以提高并行處理能力，但過多會增加資源消耗。線程優(yōu)先級：合理分配處理器的線程優(yōu)先級，確保關(guān)鍵任務(wù)優(yōu)先執(zhí)行。示例：配置線程池<threadPoolid="12345678-90ab-cdef-1234-567890abcdef">

<name>CustomThreadPool</name>

<threads>4</threads>

<priority>5</priority>

</threadPool>此配置示例展示了如何創(chuàng)建一個自定義線程池，設(shè)置4個線程和中等優(yōu)先級。6.1.3利用連接器和關(guān)系連接器使用成功和失敗關(guān)系：確保數(shù)據(jù)流中處理器的輸出關(guān)系正確配置，以處理成功和失敗的數(shù)據(jù)流。動態(tài)關(guān)系：使用動態(tài)關(guān)系處理器，如RouteOnAttribute，根據(jù)屬性動態(tài)路由數(shù)據(jù)。示例：使用RouteOnAttribute處理器<processorid="12345678-90ab-cdef-1234-567890abcdef">

<type>cessors.standard.RouteOnAttribute</type>

<name>RouteOnAttributeExample</name>

<properties>

<RoutingStrategy>Routetoasinglerelationshipbasedonanattributevalue</RoutingStrategy>

<AttributeandValuetoRouteOn>type=csv</AttributeandValuetoRouteOn>

</properties>

<relationships>

<relationshipid="12345678-90ab-cdef-1234-567890abcdef">

<name>csv</name>

<description>Routetothisrelationshipiftheattribute'type'equals'csv'</description>

</relationship>

<relationshipid="12345678-90ab-cdef-1234-567890abcdef">

<name>json</name>

<description>Routetothisrelationshipiftheattribute'type'equals'json'</description>

</relationship>

</relationships>

</processor>此配置示例展示了如何使用RouteOnAttribute處理器根據(jù)屬性值type路由數(shù)據(jù)流。6.1.4數(shù)據(jù)緩存策略緩存配置緩存策略：選擇合適的緩存策略，如Memory或Disk，以平衡性能和持久性。緩存大小：根據(jù)系統(tǒng)資源和數(shù)據(jù)量調(diào)整緩存大小。示例：配置緩存策略<controllerServiceid="12345678-90ab-cdef-1234-567890abcdef">

<type>org.apache.nifi.services.cache.StandardContentCacheControllerService</type>

<name>ContentCacheExample</name>

<properties>

<CacheStrategy>Memory</CacheStrategy>

<CacheSize>100MB</CacheSize>

</properties>

</controllerService>此配置示例展示了如何創(chuàng)建一個使用內(nèi)存緩存策略的緩存服務(wù)，設(shè)置緩存大小為100MB。6.2故障排除與最佳實(shí)踐6.2.1監(jiān)控與日志日志配置日志級別：設(shè)置處理器的日志級別，以便在出現(xiàn)問題時進(jìn)行詳細(xì)診斷。日志聚合：使用LogAggregation等控制器服務(wù)，收集和分析日志數(shù)據(jù)。示例：配置日志級別<processorid="12345678-90ab-cdef-1234-567890abcdef">

<type>cessors.standard.GetFile</type>

<name>GetFileExample</name>

<properties>

<InputDirectory>/path/to/input/directory</InputDirectory>

<FileFilter>.*\.csv</FileFilter>

<LogLevel>DEBUG</LogLevel>

</properties>

</processor>此配置示例展示了如何將GetFile處理器的日志級別設(shè)