Apache NiFi - 快速指南

---

---

Apache NiFi - 簡介

Apache NiFi 是一個功能強大、易於使用且可靠的系統，用於在不同的系統之間處理和分發資料。它基於 NSA 開發的 Niagara Files 技術，並在 8 年後捐贈給 Apache 軟體基金會。它根據 Apache 許可證版本 2.0（2004 年 1 月）分發。Apache NiFi 的最新版本為 1.7.1。

Apache NiFi 是一個即時資料攝取平臺，可以傳輸和管理不同源系統和目標系統之間的資料傳輸。它支援各種各樣的資料格式，例如日誌、地理位置資料、社交資訊流等。它還支援許多協議，例如 SFTP、HDFS 和 KAFKA 等。對各種資料來源和協議的支援使得該平臺在許多 IT 組織中廣受歡迎。

Apache NiFi - 一般特性

Apache NiFi 的一般特性如下：

• Apache NiFi 提供了一個基於 Web 的使用者介面，在設計、控制、反饋和監控之間提供了無縫的體驗。

• 它具有高度可配置性。這有助於使用者實現保證交付、低延遲、高吞吐量、動態優先順序、背壓以及在執行時修改資料流。

• 它還提供資料溯源模組來跟蹤和監控從資料流開始到結束的資料。

• 開發人員可以根據自己的需求建立自己的自定義處理器和報告任務。

• NiFi 還支援安全的協議，如 SSL、HTTPS、SSH 和其他加密方式。

• 它還支援使用者和角色管理，並且可以與 LDAP 配置以進行授權。

Apache NiFi - 關鍵概念

Apache NiFi 的關鍵概念如下：

• 處理組 - 它是一組 NiFi 資料流，幫助使用者以分層的方式管理和維護資料流。

• 資料流 - 透過連線不同的處理器來建立，以便根據需要傳輸和修改資料，從一個或多個數據源到另一個目標資料來源。

• 處理器 - 處理器是一個 Java 模組，負責從源系統獲取資料或將其儲存在目標系統中。其他處理器還用於在 FlowFile 中新增屬性或更改內容。

• FlowFile - 它是 NiFi 的基本用法，表示從源系統中提取到 NiFi 的單個數據物件。NiFi 處理器在 FlowFile 從源處理器移動到目標處理器的過程中對其進行更改。不同的事件（如建立、克隆、接收等）由資料流中的不同處理器在 FlowFile 上執行。

• 事件 - 事件表示 FlowFile 在遍歷 NiFi 資料流時發生的更改。這些事件在資料溯源中進行跟蹤。

• 資料溯源 - 它是一個儲存庫。它還有一個 UI，允許使用者檢查有關 FlowFile 的資訊，並在處理 FlowFile 期間出現任何問題時幫助進行故障排除。

Apache NiFi 優點

• Apache NiFi 透過使用 SFTP 從遠端機器獲取資料，並保證資料血緣關係。

• Apache NiFi 支援叢集，因此它可以在多個節點上執行，每個節點都使用相同的資料流處理不同的資料，從而提高資料處理效能。

• 它還提供使用者級別、處理組級別和其他模組的安全策略。

• 其 UI 也可以在 HTTPS 上執行，這使得使用者與 NiFi 的互動更加安全。

• NiFi 支援大約 188 個處理器，使用者還可以建立自定義外掛以支援各種各樣的資料系統。

Apache NiFi 缺點

• 當節點在使用者對其進行任何更改時與 NiFi 叢集斷開連線時，flow.xml 將變得無效。節點無法重新連線到叢集，除非管理員手動從連線的節點複製 flow.xml。

• 在主節點切換的情況下，Apache NiFi 存在狀態永續性問題，這有時會導致處理器無法從源系統獲取資料。

Apache NiFi - 基本概念

Apache NiFi 由一個 Web 伺服器、一個數據流控制器和一個在 Java 虛擬機器上執行的處理器組成。它還有 3 個儲存庫：FlowFile 儲存庫、內容儲存庫和溯源儲存庫，如下所示。

FlowFile 儲存庫

此儲存庫儲存透過 Apache NiFi 資料流的每個 FlowFile 的當前狀態和屬性。此儲存庫的預設位置在 Apache NiFi 的根目錄中。可以透過更改名為“nifi.flowfile.repository.directory”的屬性來更改此儲存庫的位置。

內容儲存庫

此儲存庫包含 NiFi 中所有 FlowFile 中存在的所有內容。其預設目錄也在 NiFi 的根目錄中，可以使用“org.apache.nifi.controller.repository.FileSystemRepository”屬性更改它。此目錄使用磁碟中的大量空間，因此建議在安裝磁碟中有足夠的可用空間。

溯源儲存庫

該儲存庫跟蹤並存儲在 NiFi 中流動的所有 FlowFile 的所有事件。有兩個溯源儲存庫：**易失性溯源儲存庫**（在此儲存庫中，所有溯源資料在重新啟動後都會丟失）和**永續性溯源儲存庫**。其預設目錄也在 NiFi 的根目錄中，可以使用“org.apache.nifi.provenance.PersistentProvenanceRepository”和“org.apache.nifi.provenance.VolatileProvenanceRepositor”屬性分別為相應的儲存庫更改。

Apache NiFi - 環境設定

在本章中，我們將學習 Apache NiFi 的環境設定。Apache NiFi 的安裝步驟如下：

步驟 1 - 在您的計算機上安裝當前版本的 Java。請在您的機器上設定 JAVA_HOME。您可以檢查如下所示的版本

在 Windows 作業系統 (OS) 中（使用命令提示符）：

> java -version

在 UNIX OS 中（使用終端）：

$ echo $JAVA_HOME

步驟 2 - 從 https://nifi.apache.org/download.html 下載 Apache NiFi

• 對於 Windows OS，下載 ZIP 檔案。

• 對於 UNIX OS，下載 TAR 檔案。

• 對於 Docker 映象，請訪問以下連結 https://hub.docker.com/r/apache/nifi/.

步驟 3 - Apache NiFi 的安裝過程非常簡單。過程因 OS 而異：

• Windows OS - 解壓縮 zip 包，Apache NiFi 就安裝好了。

• UNIX OS - 將 tar 檔案解壓縮到任何位置，Logstash 就安裝好了。

$tar -xvf nifi-1.6.0-bin.tar.gz

步驟 4 - 開啟命令提示符，轉到 NiFi 的 bin 目錄。例如，C:\nifi-1.7.1\bin，並執行 run-nifi.bat 檔案。

C:\nifi-1.7.1\bin>run-nifi.bat

步驟 5 - NiFi UI 啟動需要幾分鐘。使用者可以檢查 nifi-app.log，一旦 NiFi UI 啟動，使用者就可以輸入 https://:8080/nifi/ 來訪問 UI。

Apache NiFi - 使用者介面

Apache 是一個基於 Web 的平臺，使用者可以使用 Web UI 訪問。NiFi UI 非常互動式，並提供了大量有關 NiFi 的資訊。如下圖所示，使用者可以訪問以下屬性的資訊：

• 活動執行緒
• 總排隊資料
• 正在傳輸的遠端處理組
• 未傳輸的遠端處理組
• 正在執行的元件
• 已停止的元件
• 無效的元件
• 已停用的元件
• 最新的版本化處理組
• 本地修改的版本化處理組
• 過時的版本化處理組
• 本地修改且過時的版本化處理組
• 同步失敗的版本化處理組

Apache NiFi 的元件

Apache NiFi UI 包含以下元件：

處理器

使用者可以將處理器圖示拖到畫布上，併為 NiFi 中的資料流選擇所需的處理器。

輸入埠

以下圖示被拖到畫布上以將輸入埠新增到任何資料流中。

輸入埠用於從不在該處理組中的處理器獲取資料。

拖動此圖示後，NiFi 會要求輸入輸入埠的名稱，然後將其新增到 NiFi 畫布。

輸出埠

以下圖示被拖到畫布上以將輸出埠新增到任何資料流中。

輸出埠用於將資料傳輸到不在該處理組中的處理器。

拖動此圖示後，NiFi 會要求輸入輸出埠的名稱，然後將其新增到 NiFi 畫布。

處理組

使用者使用以下圖示在 NiFi 畫布中新增處理組。

拖動此圖示後，NiFi 會要求輸入處理組的名稱，然後將其新增到 NiFi 畫布。

遠端處理組

這用於在 NiFi 畫布中新增遠端處理組。

漏斗

漏斗用於將處理器的輸出傳輸到多個處理器。使用者可以使用以下圖示在 NiFi 資料流中新增漏斗。

模板

此圖示用於將資料流模板新增到 NiFi 畫布。這有助於在相同或不同的 NiFi 例項中重用資料流。

拖動後，使用者可以選擇已新增到 NiFi 中的模板。

標籤

這些用於在 NiFi 畫布上新增有關 NiFi 中任何元件的文字。它提供了一系列使用者可用於新增美觀效果的顏色。

Apache NiFi - 處理器

Apache NiFi 處理器是建立資料流的基本模組。每個處理器都有不同的功能，這有助於建立輸出 FlowFile。下圖所示的資料流使用 GetFile 處理器從一個目錄中獲取檔案，並使用 PutFile 處理器將其儲存到另一個目錄中。

GetFile

GetFile 處理器用於從特定目錄中獲取特定格式的檔案。它還為使用者提供了其他選項以更好地控制獲取過程。我們將在下面的屬性部分討論它。

GetFile 設定

以下是 GetFile 處理器的不同設定：

名稱

在“名稱”設定中，使用者可以為處理器定義任何名稱，可以根據專案或使其名稱更有意義。

啟用

使用者可以使用此設定啟用或停用處理器。

懲罰持續時間

此設定允許使用者在 FlowFile 失敗時新增懲罰時間持續時間。

產出持續時間

此設定用於指定處理器的產出時間。在此持續時間內，程序不會再次排程。

公告級別

此設定用於指定該處理器的日誌級別。

自動終止關係

此設定包含該特定流程所有可用關係的複選框列表。透過選中複選框，使用者可以對處理器進行程式設計，使其在該事件發生時終止 FlowFile，並且不將其進一步傳送到流程中。

GetFile 排程

以下是 GetFile 處理器提供的排程選項：

排程策略

您可以透過選擇“時間驅動”來按時間計劃流程，或者透過選擇“CRON 驅動”選項來指定 CRON 字串。

併發任務

此選項用於定義此處理器的併發任務排程。

執行

使用者可以使用此選項定義是在所有節點上執行處理器還是僅在主節點上執行。

執行計劃

它用於定義時間驅動策略的時間或 CRON 驅動策略的 CRON 表示式。

GetFile 屬性

GetFile 提供了多個屬性，如下面的影像所示，範圍從必填屬性（如輸入目錄和檔案過濾器）到可選屬性（如路徑過濾器和最大檔案大小）。使用者可以使用這些屬性來管理檔案獲取過程。

GetFile 註釋

此部分用於指定有關處理器的任何資訊。

PutFile

PutFile 處理器用於將檔案從資料流儲存到特定位置。

PutFile 設定

PutFile 處理器具有以下設定：

名稱

在“名稱”設定中，使用者可以根據專案或使其名稱更有意義的方式為處理器定義任何名稱。

啟用

使用者可以使用此設定啟用或停用處理器。

懲罰持續時間

此設定允許使用者在 FlowFile 失敗的情況下新增罰時持續時間。

產出持續時間

此設定用於指定處理器的產出時間。在此持續時間內，流程不會再次排程。

公告級別

此設定用於指定該處理器的日誌級別。

自動終止關係

此設定包含該特定流程所有可用關係的複選框列表。透過選中複選框，使用者可以對處理器進行程式設計，使其在該事件發生時終止 FlowFile，並且不將其進一步傳送到流程中。

PutFile 排程

以下是 PutFile 處理器提供的排程選項：

排程策略

您可以透過選擇“定時器驅動”或選擇“CRON 驅動”選項來指定 CRON 字串，從而按時間計劃流程。還有一個實驗性策略“事件驅動”，它將在特定事件發生時觸發處理器。

併發任務

此選項用於定義此處理器的併發任務排程。

執行

使用者可以使用此選項定義是在所有節點上執行處理器還是僅在主節點上執行。

執行計劃

它用於定義定時器驅動策略的時間或 CRON 驅動策略的 CRON 表示式。

PutFile 屬性

PutFile 處理器提供了一些屬性，例如“目錄”，用於指定檔案傳輸的輸出目錄，以及其他用於管理傳輸的屬性，如下面的影像所示。

PutFile 註釋

此部分用於指定有關處理器的任何資訊。

Apache NiFi - 處理器分類

在本章中，我們將討論 Apache NiFi 中的流程分類。

資料攝取處理器

資料攝取類別下的處理器用於將資料攝取到 NiFi 資料流中。這些通常是 Apache NiFi 中任何資料流的起點。屬於這些類別的一些處理器包括 GetFile、GetHTTP、GetFTP、GetKAFKA 等。

路由和中介處理器

路由和中介處理器用於根據這些 FlowFile 的屬性或內容中的資訊將 FlowFile 路由到不同的處理器或資料流。這些處理器還負責控制 NiFi 資料流。屬於此類別的一些處理器包括 RouteOnAttribute、RouteOnContent、ControlRate、RouteText 等。

資料庫訪問處理器

此資料庫訪問類別的處理器能夠從資料庫中選擇或插入資料，或執行和準備其他 SQL 語句。這些處理器主要使用 Apache NiFi 的資料連線池控制器設定。屬於此類別的一些處理器包括 ExecuteSQL、PutSQL、PutDatabaseRecord、ListDatabaseTables 等。

屬性提取處理器

屬性提取處理器負責在 NiFi 資料流中提取、分析和更改 FlowFile 屬性處理。屬於此類別的一些處理器包括 UpdateAttribute、EvaluateJSONPath、ExtractText、AttributesToJSON 等。

系統互動處理器

系統互動處理器用於在任何作業系統中執行程序或命令。這些處理器還以多種語言執行指令碼，以與各種系統互動。屬於此類別的一些處理器包括 ExecuteScript、ExecuteProcess、ExecuteGroovyScript、ExecuteStreamCommand 等。

資料轉換處理器

屬於資料轉換的處理器能夠更改 FlowFile 的內容。這些可以用於完全替換 FlowFile 的資料，通常在使用者必須將 FlowFile 作為 HTTP 主體傳送以呼叫 InvokeHTTP 處理器時使用。屬於此類別的一些處理器包括 ReplaceText、JoltTransformJSON 等。

傳送資料處理器

傳送資料處理器通常是資料流中的最終處理器。這些處理器負責將資料儲存或傳送到目標伺服器。在成功儲存或傳送資料後，這些處理器將 FlowFile 丟棄，並使用成功關係。屬於此類別的一些處理器包括 PutEmail、PutKafka、PutSFTP、PutFile、PutFTP 等。

拆分和聚合處理器

這些處理器用於拆分和合並 FlowFile 中存在的內容。屬於此類別的一些處理器包括 SplitText、SplitJson、SplitXml、MergeContent、SplitContent 等。

HTTP 處理器

這些處理器處理 HTTP 和 HTTPS 呼叫。屬於此類別的一些處理器包括 InvokeHTTP、PostHTTP、ListenHTTP 等。

AWS 處理器

AWS 處理器負責與 Amazon Web Services 系統互動。屬於此類別的一些處理器包括 GetSQS、PutSNS、PutS3Object、FetchS3Object 等。

Apache NiFi - 處理器關係

在 Apache NiFi 資料流中，FlowFile 透過連線從一個處理器移動到另一個處理器，這些連線使用處理器之間的關係進行驗證。每當建立連線時，開發人員都會選擇這些處理器之間的一種或多種關係。

如上圖所示，黑色矩形中的複選框是關係。如果開發人員選中這些複選框，則當關系成功或失敗或兩者兼有時，FlowFile 將在該特定處理器中終止。

成功

當處理器成功處理 FlowFile（例如從任何資料來源儲存或獲取資料，而不會遇到任何連線、身份驗證或任何其他錯誤）時，FlowFile 將進入成功關係。

失敗

當處理器無法在沒有錯誤的情況下處理 FlowFile（例如身份驗證錯誤或連線問題等）時，FlowFile 將進入失敗關係。

開發人員還可以使用連線將 FlowFile 傳輸到其他處理器。開發人員可以選擇並對其進行負載平衡，但負載平衡僅在 1.8 版中釋出，本教程中不會介紹。

如上圖所示，紅色標記的連線具有失敗關係，這意味著所有帶有錯誤的 FlowFile 將進入左側的處理器，並且所有沒有錯誤的 FlowFile 將分別傳輸到綠色標記的連線。

現在讓我們繼續討論其他關係。

comms.failure

當由於通訊故障無法從遠端伺服器獲取 FlowFile 時，將滿足此關係。

not.found

對於我們從遠端伺服器收到“未找到”訊息的任何 FlowFile，都將移動到**not.found**關係。

permission.denied

當 NiFi 由於許可權不足而無法從遠端伺服器獲取 FlowFile 時，它將透過此關係移動。

Apache NiFi - FlowFile

FlowFile 是 Apache NiFi 中的基本處理實體。它包含資料內容和屬性，NiFi 處理器使用這些內容和屬性來處理資料。檔案內容通常包含從源系統獲取的資料。Apache NiFi FlowFile 的最常見屬性包括：

UUID

它代表通用唯一識別符號，是 NiFi 生成的 FlowFile 的唯一標識。

檔名

此屬性包含該 FlowFile 的檔名，並且不應包含任何目錄結構。

檔案大小

它包含 Apache NiFi FlowFile 的大小。

mime.type

它指定此 FlowFile 的 MIME 型別。

path

此屬性包含 FlowFile 所屬檔案的相對路徑，並且不包含檔名。

Apache NiFi - 佇列

Apache NiFi 資料流連線具有一個排隊系統來處理大量資料流入。這些佇列可以處理非常大量的 FlowFile，以允許處理器按順序處理它們。

上圖中的佇列具有 1 個透過成功關係傳輸的 FlowFile。使用者可以透過選擇下拉列表中的“列出佇列”選項來檢查 FlowFile。在任何過載或錯誤的情況下，使用者還可以透過選擇“清空佇列”選項來清除佇列，然後使用者可以重新啟動流程以再次獲取資料流中的這些檔案。

佇列中的 FlowFile 列表包含位置、UUID、檔名、檔案大小、佇列持續時間和血緣持續時間。使用者可以透過單擊 FlowFile 列表第一列中存在的 info 圖示來檢視 FlowFile 的所有屬性和內容。

Apache NiFi - 處理組

在 Apache NiFi 中，使用者可以在不同的流程組中維護不同的資料流。這些組可以基於 Apache NiFi 例項支援的不同專案或組織。

NiFi UI 頂部選單中的第四個符號（如上圖所示）用於在 NiFi 畫布中新增流程組。“Tutorialspoint.com_ProcessGroup”流程組包含一個數據流，其中目前有四個處理器處於停止狀態，如上圖所示。可以以分層方式建立流程組，以便以更易於理解的結構管理資料流。

在 NiFi UI 的頁尾中，您可以看到流程組，並且可以返回到使用者當前所在的流程組的頂部。

要檢視 NiFi 中存在的流程組的完整列表，使用者可以使用 NiFi UI 左側頂部存在的選單轉到摘要。在摘要中，有一個流程組選項卡，其中列出了所有流程組，以及版本狀態、傳輸/大小、輸入/大小、讀取/寫入、輸出/大小等引數，如下面的圖片所示。

Apache NiFi - 標籤

Apache NiFi 提供標籤，使開發人員能夠編寫有關 NiFi 畫布中存在的元件的資訊。NiFi UI 頂部選單中最左邊的圖示用於在 NiFi 畫布中新增標籤。

開發人員可以透過右鍵單擊標籤並從選單中選擇適當的選項來更改標籤的顏色和文字大小。

Apache NiFi - 配置

Apache NiFi 是一個高度可配置的平臺。conf 目錄中的 nifi.properties 檔案

包含大部分配置。

Apache NiFi 的常用屬性如下：

核心屬性

此部分包含執行 NiFi 例項所必需的屬性。

狀態管理

這些屬性用於儲存元件的狀態，有助於啟動處理，其中元件在重新啟動後以及在下一個計劃執行中保留。

FlowFile 儲存庫

現在讓我們瞭解 FlowFile 儲存庫的重要細節 -

Apache NiFi - 管理

Apache NiFi 提供對 ambari、zookeeper 等多種工具的支援，用於管理目的。NiFi 還在 nifi.properties 檔案中提供配置，以便為管理員設定 HTTPS 和其他內容。

zookeeper

NiFi 本身不處理叢集中的投票過程。這意味著當建立叢集時，所有節點都是主節點和協調器。因此，配置 ZooKeeper 以管理主節點和協調器的投票。nifi.properties 檔案包含一些用於設定 ZooKeeper 的屬性。

啟用 HTTPS

要透過 HTTPS 使用 NiFi，管理員必須生成金鑰庫和信任庫，並在 nifi.properties 檔案中設定一些屬性。TLS 工具包可用於生成啟用 Apache NiFi 中 HTTPS 所需的所有金鑰。

其他管理屬性

還有一些其他屬性，管理員使用這些屬性來管理 NiFi 及其服務連續性。

Apache NiFi - 建立資料流

Apache NiFi 提供大量元件，以幫助開發人員為任何型別的協議或資料來源建立資料流。要建立流，開發人員會將元件從選單欄拖動到畫布上，然後透過單擊並從一個元件拖動到另一個元件來連線它們。

通常，NiFi 在流的開頭有一個偵聽器元件，例如 getfile，它從源系統獲取資料。在另一端有一個發射器元件，例如 putfile，並且中間有一些元件來處理資料。

例如，讓我們建立一個流，它從一個目錄獲取一個空檔案，在該檔案中新增一些文字，然後將其放入另一個目錄。

• 首先，將處理器圖示拖到 NiFi 畫布上，然後從列表中選擇 GetFile 處理器。

• 建立一個輸入目錄，例如 **c:\inputdir**。

• 右鍵單擊處理器並選擇配置，在屬性選項卡中新增 **輸入目錄 (c:\inputdir)** 並單擊應用，然後返回畫布。

• 將處理器圖示拖到畫布上，然後從列表中選擇 ReplaceText 處理器。

• 右鍵單擊處理器並選擇配置。在**屬性**選項卡中，在替換值的文字框中新增一些文字，例如**“Hello tutorialspoint.com”**，然後單擊應用。

• 轉到設定選項卡，選中右側的失敗複選框，然後返回畫布。

• 將 GetFIle 處理器連線到 ReplaceText 的成功關係。

• 將處理器圖示拖到畫布上，然後從列表中選擇 PutFile 處理器。

• 建立一個輸出目錄，例如 **c:\outputdir**。

• 右鍵單擊處理器並選擇配置。在屬性選項卡中，新增 **目錄 (c:\outputdir)** 並單擊應用，然後返回畫布。

• 轉到設定選項卡，選中右側的失敗和成功複選框，然後返回畫布。

• 將 ReplaceText 處理器連線到 PutFile 的成功關係。

• 現在啟動流並在輸入目錄中新增一個空檔案，您將看到它將移動到輸出目錄，並且文字將新增到該檔案中。

透過遵循上述步驟，開發人員可以選擇任何處理器和其他 NiFi 元件來為其組織或客戶建立合適的流。

Apache NiFi - 模板

Apache NiFi 提供了模板的概念，這使得更容易重用和分發 NiFi 流。其他開發人員或其他 NiFi 叢集可以使用這些流。它還有助於 NiFi 開發人員在 GitHub 等儲存庫中共享他們的工作。

建立模板

讓我們為我們在第 15 章“Apache NiFi - 建立流”中建立的流建立一個模板。

使用 Shift 鍵選擇流的所有元件，然後單擊 NiFi 畫布左側的建立模板圖示。您還可以看到一個工具箱，如上圖所示。單擊上面圖片中藍色標記的**建立模板**圖示。輸入模板的名稱。開發人員還可以新增描述，這是可選的。

下載模板

然後轉到 NiFi UI 右上角選單中的 NiFi 模板選項，如下面的圖片所示。

現在單擊要下載的模板的下載圖示（位於列表右側）。將下載一個帶有模板名稱的 XML 檔案。

上傳模板

要在 NiFi 中使用模板，開發人員必須使用 UI 將其 xml 檔案上傳到 NiFi。在建立模板圖示旁邊有一個上傳模板圖示（在下面的圖片中用藍色標記），單擊該圖示並瀏覽 xml。

新增模板

在 NiFi UI 的頂部工具欄中，模板圖示位於標籤圖示之前。該圖示在下面的圖片中用藍色標記。

拖動模板圖示並從下拉列表中選擇模板，然後單擊新增。它會將模板新增到 NiFi 畫布。

Apache NiFi - API

NiFi 提供大量 API，幫助開發人員從任何其他工具或自定義開發的應用程式中更改和獲取 NiFi 的資訊。在本教程中，我們將使用 Google Chrome 中的 postman 應用程式來解釋一些示例。

要將 postman 新增到您的 Google Chrome，請訪問下面提到的 URL 並單擊新增到 Chrome 按鈕。您現在將看到一個新應用新增到您的 Google Chrome。

chrome 網上應用商店

NiFi rest API 的當前版本是 1.8.0，文件位於下面提到的 URL 中。

https://nifi.apache.org/docs/nifi-docs/rest-api/index.html

以下是最常用的 NiFi rest API 模組 -

• http://<nifi url>:<nifi port>/nifi-api/<api-path>

• 如果啟用了 HTTPS，則為 https://<nifi url>:<nifi port>/nifi-api/<api-path>

現在讓我們考慮一個示例並在 postman 上執行以獲取有關正在執行的 NiFi 例項的詳細資訊。

請求

GET https://:8080/nifi-api/flow/about

響應

{
   "about": {
      "title": "NiFi",
      "version": "1.7.1",
      "uri": "https://:8080/nifi-api/",
      "contentViewerUrl": "../nifi-content-viewer/",
      "timezone": "SGT",
      "buildTag": "nifi-1.7.1-RC1",
      "buildTimestamp": "07/12/2018 12:54:43 SGT"
   }
}

Apache NiFi - 資料溯源

Apache NiFi 記錄並存儲資料流中所有攝取資料事件的資訊。資料溯源儲存庫儲存這些資訊，並提供 UI 用於搜尋這些事件資訊。資料溯源可以訪問 NiFi 全域性級別和處理器級別。

下表列出了 NiFi 資料溯源事件列表中的不同欄位：

要獲取有關事件的更多資訊，使用者可以單擊 NiFi 資料溯源 UI 第一列中顯示的資訊圖示。

nifi.properties 檔案中有一些屬性用於管理 NiFi 資料溯源儲存庫。

Apache NiFi - 監控

在 Apache NiFi 中，有多種方法可以監控系統的不同統計資訊，例如錯誤、記憶體使用情況、CPU 使用情況、資料流統計資訊等。本教程將討論其中最常用的幾種。

內建監控

在本節中，我們將詳細瞭解 Apache NiFi 中的內建監控。

公告板

公告板即時顯示 NiFi 處理器生成的最新錯誤和警告。要訪問公告板，使用者需要轉到右側的向下選單並選擇“公告板”選項。它會自動重新整理，使用者也可以停用它。使用者還可以透過雙擊錯誤導航到實際的處理器。使用者還可以透過以下方式過濾公告：

• 按訊息
• 按名稱
• 按 ID
• 按組 ID

資料溯源 UI

要監控任何特定處理器或整個 NiFi 中發生的事件，使用者可以從與公告板相同的選單訪問資料溯源。使用者還可以透過以下欄位過濾資料溯源儲存庫中的事件：

• 按元件名稱
• 按元件型別
• 按型別

NiFi 摘要 UI

Apache NiFi 摘要也可以從與公告板相同的選單訪問。此 UI 包含有關該特定 NiFi 例項或叢集的所有元件的資訊。它們可以透過名稱、型別或 URI 進行過濾。不同元件型別有不同的選項卡。以下是可以監控的 NiFi 摘要 UI 中的元件：

• 處理器
• 輸入埠
• 輸出埠
• 遠端程序組
• 連線
• 程序組

在此 UI 中，在右下角有一個名為“系統診斷”的連結，用於檢查 JVM 統計資訊。

報告任務

Apache NiFi 提供多個報告任務來支援外部監控系統，例如 Ambari、Grafana 等。開發人員可以建立自定義報告任務，也可以配置內建任務將 NiFi 的指標傳送到外部監控系統。下表列出了 NiFi 1.7.1 提供的報告任務。

NiFi API

有一個名為“系統診斷”的 API，可用於在任何自定義開發的應用程式中監控 NiFi 統計資訊。讓我們在 Postman 中檢查此 API。

請求

https://:8080/nifi-api/system-diagnostics

響應

{
   "systemDiagnostics": {
      "aggregateSnapshot": {
         "totalNonHeap": "183.89 MB",
         "totalNonHeapBytes": 192819200,
         "usedNonHeap": "173.47 MB",
         "usedNonHeapBytes": 181894560,
         "freeNonHeap": "10.42 MB",
         "freeNonHeapBytes": 10924640,
         "maxNonHeap": "-1 bytes",
         "maxNonHeapBytes": -1,
         "totalHeap": "512 MB",
         "totalHeapBytes": 536870912,
         "usedHeap": "273.37 MB",
         "usedHeapBytes": 286652264,
         "freeHeap": "238.63 MB",
         "freeHeapBytes": 250218648,
         "maxHeap": "512 MB",
         "maxHeapBytes": 536870912,
         "heapUtilization": "53.0%",
         "availableProcessors": 4,
         "processorLoadAverage": -1,
         "totalThreads": 71,
         "daemonThreads": 31,
         "uptime": "17:30:35.277",
         "flowFileRepositoryStorageUsage": {
            "freeSpace": "286.93 GB",
            "totalSpace": "464.78 GB",
            "usedSpace": "177.85 GB",
            "freeSpaceBytes": 308090789888,
            "totalSpaceBytes": 499057160192,
            "usedSpaceBytes": 190966370304,
            "utilization": "38.0%"
         },
         "contentRepositoryStorageUsage": [
            {
               "identifier": "default",
               "freeSpace": "286.93 GB",
               "totalSpace": "464.78 GB",
               "usedSpace": "177.85 GB",
               "freeSpaceBytes": 308090789888,
               "totalSpaceBytes": 499057160192,
               "usedSpaceBytes": 190966370304,
               "utilization": "38.0%"
            }
         ],
         "provenanceRepositoryStorageUsage": [
            {
               "identifier": "default",
               "freeSpace": "286.93 GB",
               "totalSpace": "464.78 GB",
               "usedSpace": "177.85 GB",
               "freeSpaceBytes": 308090789888,
               "totalSpaceBytes": 499057160192,
               "usedSpaceBytes": 190966370304,
               "utilization": "38.0%"
            }
         ],
         "garbageCollection": [
            {
               "name": "G1 Young Generation",
               "collectionCount": 344,
               "collectionTime": "00:00:06.239",
               "collectionMillis": 6239
            },
            {
               "name": "G1 Old Generation",
               "collectionCount": 0,
               "collectionTime": "00:00:00.000",
               "collectionMillis": 0
            }
         ],
         "statsLastRefreshed": "09:30:20 SGT",
         "versionInfo": {
            "niFiVersion": "1.7.1",
            "javaVendor": "Oracle Corporation",
            "javaVersion": "1.8.0_151",
            "osName": "Windows 7",
            "osVersion": "6.1",
            "osArchitecture": "amd64",
            "buildTag": "nifi-1.7.1-RC1",
            "buildTimestamp": "07/12/2018 12:54:43 SGT"
         }
      }
   }
}

Apache NiFi - 升級

在開始升級 Apache NiFi 之前，請閱讀發行說明以瞭解更改和新增功能。使用者需要評估這些新增功能和更改對其當前 NiFi 安裝的影響。以下是從 Apache NiFi 新版本獲取發行說明的連結：

https://cwiki.apache.org/confluence/display/NIFI/Release+Notes

在叢集設定中，使用者需要升級叢集中每個節點的 NiFi 安裝。請按照以下步驟升級 Apache NiFi。

• 備份當前 NiFi 中存在的 lib 或任何其他資料夾中的所有自定義 NAR。

• 下載新版本的 Apache NiFi。以下是從最新 NiFi 版本下載原始碼和二進位制檔案的連結。

  https://nifi.apache.org/download.html

• 在當前 NiFi 的同一安裝目錄中建立一個新目錄，並將新版本的 Apache NiFi 解壓縮到該目錄。

• 優雅地停止 NiFi。首先停止所有處理器，並讓資料流中存在的 FlowFile 都被處理完成。一旦沒有更多 FlowFile，停止 NiFi。

• 將當前 NiFi 安裝的 authorizers.xml 配置複製到新版本。

• 從當前版本更新新 NiFi 版本的 bootstrap-notification-services.xml 和 bootstrap.conf 中的值。

• 將 logback.xml 中的自定義日誌記錄新增到新 NiFi 安裝中。

• 從當前版本配置 login-identity-providers.xml 中的登入身份提供程式。

• 從當前版本更新新 NiFi 安裝的 nifi.properties 中的所有屬性。

• 請確保新版本的組和使用者與當前版本相同，以避免任何許可權被拒絕錯誤。

• 將當前版本的 state-management.xml 配置複製到新版本。

• 將以下目錄的內容從當前版本的 NiFi 安裝複製到新版本中的相同目錄。

  • ./conf/flow.xml.gz

  • 以及 archive 目錄中的 flow.xml.gz。

  • 對於溯源和內容儲存庫，請在 nifi.properties 檔案中將值更改為當前儲存庫。

  • 從 ./state/local 複製狀態，或如果指定了任何其他外部目錄，則在 nifi.properties 中進行更改。

• 重新檢查所有執行的更改，並檢查它們是否對新 NiFi 版本中新增的任何新更改有影響。如果有任何影響，請檢查解決方案。

• 啟動所有 NiFi 節點，並驗證所有資料流是否正常工作，儲存庫是否正在儲存資料，並且 UI 是否在沒有錯誤的情況下檢索資料。

• 監控公告板一段時間，以檢查是否有任何新的錯誤。

• 如果新版本正常工作，則可以將當前版本存檔並從目錄中刪除。

Apache NiFi - 遠端處理組

Apache NiFi 遠端程序組 (RPG) 使資料流能夠使用站點到站點協議將資料流中的 FlowFile 指向不同的 NiFi 例項。截至 1.7.1 版本，NiFi 不提供負載均衡關係，因此 RPG 用於在 NiFi 資料流中進行負載均衡。

開發人員可以透過將 NiFi UI 頂部工具欄中的圖示拖動到畫布上（如上圖所示）來新增 RPG。要配置 RPG，開發人員必須新增以下欄位：

開發人員需要在使用之前啟用它，就像我們使用處理器之前啟動它們一樣。

Apache NiFi - 控制器設定

Apache NiFi 提供共享服務，處理器和報告任務可以使用這些服務，稱為控制器設定。這些就像資料庫連線池，可以被訪問相同資料庫的處理器使用。

要訪問控制器設定，請使用 NiFi UI 右上角的下拉選單，如下面的影像所示。

Apache NiFi 提供了許多控制器設定，我們將討論一個常用的設定以及如何在 NiFi 中進行設定。

DBCPConnectionPool

單擊“控制器設定”選項後，在 NiFi 設定頁面中新增加號。然後從控制器設定列表中選擇 DBCPConnectionPool。DBCPConnectionPool 將新增到主 NiFi 設定頁面，如下面的影像所示。

它包含以下有關控制器 **設定：名稱** 的資訊：

• 型別
• 捆綁包
• 狀態
• 範圍
• 配置和刪除圖示

單擊“配置”圖示並填寫所需欄位。這些欄位列在下表中：

要停止或配置控制器設定，首先必須停止所有附加的 NiFi 元件。NiFi 還為控制器設定添加了範圍以管理其配置。因此，只有共享相同設定的元件不會受到影響，並將使用相同的控制器設定。

Apache NiFi - 報告任務

Apache NiFi 報告任務類似於在後臺執行併發送或記錄 NiFi 例項統計資訊的控制器服務。NiFi 報告任務也可以從與控制器設定相同的頁面訪問，但在不同的選項卡中。

要新增報告任務，開發人員需要點選報告任務頁面右上角的加號按鈕。這些報告任務主要用於監控 NiFi 例項的活動，無論是公告還是來源。主要這些報告任務使用站點到站點將 NiFi 統計資料傳輸到其他節點或外部系統。

現在讓我們新增一個已配置的報告任務以瞭解更多資訊。

MonitorMemory

此報告任務用於在記憶體池超過指定百分比時生成公告。請按照以下步驟配置 MonitorMemory 報告任務 -

• 新增加號並在列表中搜索 MonitorMemory。

• 選擇 MonitorMemory 並點選新增。

• 一旦它新增到報告任務主頁面，點選配置圖示。

• 在屬性選項卡中，選擇要監控的記憶體池。

• 選擇希望公告提醒使用者的百分比。

• 啟動報告任務。

Apache NiFi - 自定義處理器

Apache NiFi 是一個開源平臺，允許開發人員在 NiFi 庫中新增自定義處理器。請按照以下步驟建立自定義處理器。

• 從下面給出的連結下載 Maven 最新版本。

  https://maven.apache.org/download.cgi

• 新增名為 M2_HOME 的環境變數，並將值設定為 maven 的安裝目錄。

• 從以下連結下載 Eclipse IDE。

  https://www.eclipse.org/downloads/download.php

• 開啟命令提示符並執行 Maven Archetype 命令。

> mvn archetype:generate

• 在原型專案中搜索 nifi 型別。

• 選擇 org.apache.nifi:nifi-processor-bundle-archetype 專案。

• 然後從版本列表中選擇最新版本，即本教程的 1.7.1。

• 輸入 groupId、artifactId、version、package 和 artifactBaseName 等。

• 然後將建立一個包含兩個目錄的 Maven 專案。

  • nifi-<artifactBaseName>-processors

  • nifi-<artifactBaseName>-nar

• 在 nifi-<artifactBaseName>-processors 目錄中執行以下命令以將專案新增到 eclipse 中。

mvn install eclipse:eclipse

• 開啟 eclipse 並從檔案選單中選擇匯入。

• 然後選擇“將現有專案匯入工作區”，並將專案從 nifi-<artifactBaseName>-processors 目錄新增到 eclipse 中。

• 在 public void onTrigger(ProcessContext context, ProcessSession session) 函式中新增您的程式碼，該函式在每次計劃處理器執行時執行。

• 然後透過執行下面提到的命令將程式碼打包到 NAR 檔案中。

mvn clean install

• NAR 檔案將在 nifi- 中建立-nar/target 目錄。

• 將 NAR 檔案複製到 Apache NiFi 的 lib 資料夾並重新啟動 NiFi。

• NiFi 成功重啟後，檢查處理器列表中是否有新的自定義處理器。

• 對於任何錯誤，請檢查 ./logs/nifi.log 檔案。

Apache NiFi - 自定義控制器服務

Apache NiFi 是一個開源平臺，允許開發人員在 Apache NiFi 中新增自定義控制器服務。步驟和工具與建立自定義處理器幾乎相同。

• 開啟命令提示符並執行 Maven Archetype 命令。

> mvn archetype:generate

• 在原型專案中搜索 nifi 型別。

• 選擇org.apache.nifi:nifi-service-bundle-archetype專案。

• 然後從版本列表中選擇最新版本 - 本教程的 1.7.1。

• 輸入 groupId、artifactId、version、package 和 artifactBaseName 等。

• 將建立一個包含目錄的 Maven 專案。

  • nifi-<artifactBaseName>

  • nifi-<artifactBaseName>-nar

  • nifi-<artifactBaseName>-api

  • nifi-<artifactBaseName>-api-nar

• 在 nifi-<artifactBaseName> 和 nifi-<artifactBaseName>-api 目錄中執行以下命令以將這兩個專案新增到 eclipse 中。

  • mvn install eclipse:eclipse

• 開啟 eclipse 並從檔案選單中選擇匯入。

• 然後選擇“將現有專案匯入工作區”，並將專案從 nifi-<artifactBaseName> 和 nifi-<artifactBaseName>-api 目錄新增到 eclipse 中。

• 在原始檔中新增您的程式碼。

• 然後透過執行下面提到的命令將程式碼打包到 NAR 檔案中。

  • mvn clean install

• 將在每個 nifi-<artifactBaseName>/target 和 nifi-<artifactBaseName>-api/target 目錄中建立兩個 NAR 檔案。

• 將這些 NAR 檔案複製到 Apache NiFi 的 lib 資料夾並重新啟動 NiFi。

• NiFi 成功重啟後，檢查處理器列表中是否有新的自定義處理器。

• 對於任何錯誤，請檢查./logs/nifi.log檔案。

Apache NiFi - 日誌

Apache NiFi 使用 logback 庫來處理其日誌記錄。NiFi 的 conf 目錄中有一個名為 logback.xml 的檔案，用於配置 NiFi 中的日誌記錄。日誌生成在 NiFi 的 logs 資料夾中，日誌檔案如下所述。

nifi-app.log

這是 nifi 的主日誌檔案，記錄了從 NAR 檔案載入到 NiFi 元件遇到的執行時錯誤或公告等所有 apache NiFi 應用程式活動。以下是logback.xml檔案中nifi-app.log檔案的預設附加程式。

<appender name="APP_FILE"
class="ch.qos.logback.core.rolling.RollingFileAppender">
   <file>${org.apache.nifi.bootstrap.config.log.dir}/nifi-app.log</file>
   <rollingPolicy
      class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
      <fileNamePattern>
         ${org.apache.nifi.bootstrap.config.log.dir}/
	      nifi-app_%d{yyyy-MM-dd_HH}.%i.log
      </fileNamePattern>
      <maxFileSize>100MB</maxFileSize>
      <maxHistory>30</maxHistory>
   </rollingPolicy>
   <immediateFlush>true</immediateFlush>
   <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
   </encoder>
</appender>

附加程式名稱為 APP_FILE，類為 RollingFileAppender，這意味著記錄器使用回滾策略。預設情況下，最大檔案大小為 100 MB，可以更改為所需大小。APP_FILE 的最大保留時間為 30 個日誌檔案，可以根據使用者需求進行更改。

nifi-user.log

此日誌包含使用者事件，如 Web 安全、Web API 配置、使用者授權等。以下是 logback.xml 檔案中 nifi-user.log 的附加程式。

<appender name="USER_FILE"
   class="ch.qos.logback.core.rolling.RollingFileAppender">
   <file>${org.apache.nifi.bootstrap.config.log.dir}/nifi-user.log</file>
   <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>
         ${org.apache.nifi.bootstrap.config.log.dir}/
	      nifi-user_%d.log
      </fileNamePattern>
      <maxHistory>30</maxHistory>
   </rollingPolicy>
   <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
   </encoder>
</appender>

附加程式名稱為 USER_FILE。它遵循回滾策略。USER_FILE 的最大保留期為 30 個日誌檔案。以下是 nifi-user.log 中存在的 USER_FILE 附加程式的預設記錄器。

<logger name="org.apache.nifi.web.security" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.web.api.config" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.authorization" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.cluster.authorization" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>
<logger name="org.apache.nifi.web.filter.RequestLogger" level="INFO" additivity="false">
   <appender-ref ref="USER_FILE"/>
</logger>

nifi-bootstrap.log

此日誌包含引導日誌、apache NiFi 的標準輸出（程式碼中編寫的全部 system.out，主要用於除錯）和標準錯誤（程式碼中編寫的全部 system.err）。以下是 logback.log 中 nifi-bootstrap.log 的預設附加程式。

<appender name="BOOTSTRAP_FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
   <file>${org.apache.nifi.bootstrap.config.log.dir}/nifi-bootstrap.log</file>
   <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
      <fileNamePattern>
         ${org.apache.nifi.bootstrap.config.log.dir}/nifi-bootstrap_%d.log
      </fileNamePattern>
      <maxHistory>5</maxHistory>
   </rollingPolicy>
   <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
      <pattern>%date %level [%thread] %logger{40} %msg%n</pattern>
   </encoder>
</appender>

nifi-bootstrap.log 檔案的附加程式名稱為 BOOTSTRAP_FILE，它也遵循回滾策略。BOOTSTRAP_FILE 附加程式的最大保留時間為 5 個日誌檔案。以下是 nifi-bootstrap.log 檔案的預設記錄器。

<logger name="org.apache.nifi.bootstrap" level="INFO" additivity="false">
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>
<logger name="org.apache.nifi.bootstrap.Command" level="INFO" additivity="false">
   <appender-ref ref="CONSOLE" />
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>
<logger name="org.apache.nifi.StdOut" level="INFO" additivity="false">
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>
<logger name="org.apache.nifi.StdErr" level="ERROR" additivity="false">
   <appender-ref ref="BOOTSTRAP_FILE" />
</logger>