近日，StreamNative 和 Cloudera 官方宣佈共同開源 Apache NiFi 和 Apache Pulsar 聯合解決方案，將二者整合打造成完整的邊緣到雲資料的流平臺。

StreamNative 由 Apache Pulsar 創始團隊組建，專注於 Apache Pulsar 社群和生態構建，並圍繞 Apache Pulsar 打造批流一體的雲原生解決方案；Cloudera 團隊包括一些 Apache NiFi 的原始開發人員，並通過 Apache NiFi 打造資料流。通過將 NiFi 與 Pulsar 整合，企業能夠建立一個雲原生、可擴充套件的實時流資料平臺，來攝取、轉換和分析海量資料。

本文將介紹該處理器的開源背景，以及如何通過簡單配置設定 Apache NiFi 來大規模地生產消費來自 Pulsar 主題的訊息。Cloudera 為 Data Hub 7.2.14 和更高版本的 CDF^[1] 提供開箱即用的處理器。

關於 Apache NiFi

Apache NiFi^[2] 初始的專案名稱是 Niagara Files，是由美國國家安全域性(NSA)貢獻給 Apache 軟體基金會的開源專案，其設計初衷是自動化系統間的資料流。2015 年 7 月，NiFi 從 Apache 軟體基金會畢業併成為 Apache 軟體基金會的頂級專案。

NiFi 實現了基於流程式設計的一種視覺化工具，使用者可以通過 NiFi 構建將資料從一個平臺（如資料庫、雲端儲存和訊息系統）移動到另一個平臺的資料流。

NiFi 可幫助使用者自動在不同資料來源和系統之間移動資料，保證資料攝取的快速、簡單和安全；NiFi 提供實時控制來輕鬆管理任何來源和任何目的地之間的資料移動；它還提供事件級別的資料溯源和可追溯性，使用者可以將每條資料追溯到其源頭。

NiFi 平臺包含 100 多個預構建處理器的集合，可用於從資料來源向資料目的地對資料進行豐富、路由等轉換。

關於 Apache Pulsar

Apache Pulsar 是雲原生時代訊息佇列和流融合系統，提供統一的消費模型，支援訊息佇列和流兩種場景，既能為佇列場景提供企業級讀寫服務質量和強一致性保障，又能為流場景提供高吞吐、低延遲；採用儲存計算分離架構，支援大叢集、多租戶、百萬級 Topic、跨地域資料複製、持久化儲存、分層儲存、高可擴充套件性等企業級和金融級功能。

Pulsar 的核心是使用複製的分散式 ledger 來提供持久的流儲存，保證輕鬆擴充套件以保留 PB 級的資料。Pulsar 的可擴充套件流儲存使其成為事件資料的完美長期儲存庫。通過其訊息保留策略，使用者可以無限期地保留歷史事件資料，方便在未來隨時對事件資料進行流式分析。

處理器：將 Apache Pulsar 與 Apache NiFi 互補

Apache NiFi 和 Apache Pulsar 的功能在現代流資料架構中相互補充。NiFi 提供了一種資料流解決方案，可自動執行軟體系統之間的資料流。因此，它可以充當不同資料來源之間的短期緩衝區，而不是長期的資料儲存庫。

相反，Pulsar 旨在充當事件資料的長期儲存庫，並提供與常見的流處理框架（如 Flink 和 Spark）的強大整合。通過結合這兩種技術，使用者可以建立一個強大的實時資料處理和分析平臺。

這些技術結合所實現的協同效應將在資料平臺中得到顯著體現。NiFi 提供了使用者所有的資料流管理需求，包括優先順序、背壓和邊緣智慧。

使用者可以使用 NiFi 廣泛的聯結器套件來自動將資料流到訊息流平臺，同時執行 ETL 處理。資料轉換後，可以通過這些專為 Apache Pulsar 設計的 NiFi 處理器直接路由到 Pulsar 的持久流儲存，以便長期保留。

一旦資料儲存在 Pulsar 中，就可以隨時供如 Flink 或 Spark 等各種常見的流處理引擎使用，將資料用於更復雜的流處理和分析場景。

簡而言之，NiFi 豐富的聯結器允許使用者輕鬆地將資料“輸入”到訊息流平臺，同時保證 Pulsar 與 Flink 或者 Spark 的整合可以輕鬆獲得實時洞察。

Apache Pulsar 和 Apache NiFi 的結合建立了一個完整的邊緣到雲資料的流平臺，可跨多個應用程式提供實時洞察。該整合適用於多個行業和場景，舉例來說，在網路安全行業，使用者需要儘快識別和檢測威脅，要求系統具有攝取和解析日誌資料的能力；製造業、採礦業以及石油和天然氣等眾多行業都需要能夠從不同位置攝取大量 IoT 感測器資料，企業需要近實時地分析這些海量資料，以防止災難性的裝置故障和/或防止可能導致的運營突然中斷；在金融服務行業，演算法交易或加密貨幣套利等時間敏感型應用要求系統具有近實時地攝取和處理資料的能力。

視訊演示

接下來讓我們來看看這些處理器的實際應用。本視訊演示了配置和使用這些處理器向 Apache Pulsar 叢集傳送資料並從其接收資料的過程。

視訊演示：

從視訊演示中可以看到，一共有四個處理器：PublishPulsar 和 PublishPulsarRecord 用於向 Pulsar 釋出資料；ConsumePulsar 和 ConsumePulsarRecord 用於消費來自 Pulsar 的資料。bundle 中還包含兩個控制器服務：一個用於建立 Pulsar 客戶端，另一個用於身份驗證以保護 Pulsar 叢集。

使用處理器

這些處理器在公有云上的 CDF 7.2.14 版本及以上版本可用，參考文件^[3]。如果您希望在其他 Apache NiFi 叢集中使用這些處理器，可以直接從 Maven 中央程式碼倉庫^[4]下載工件，或者直接通過原始碼構建^[5]。

相關閱讀

• • Pulsar Summit 演講視訊: 邊緣 AI 場景中 FLiPN 技術棧(Flink, NiFi, Pulsar)的應用^[6]

• • 下載演示程式碼^[7] 開始執行處理器。

• • Producing and Consuming Pulsar messages with Apache NiFi^[8]

• • FLiP-Transit GitHub 倉庫^[9]

• • FLiPN-Demos GitHub 倉庫^[10]

關於 StreamNative

StreamNative 是一家開源基礎軟體公司，由 Apache 軟體基金會頂級專案 Apache Pulsar 創始團隊組建而成，圍繞 Pulsar 打造下一代雲原生批流融合資料平臺。StreamNative 作為 Apache Pulsar 商業化公司，專注於開源生態和社群構建，致力於前沿技術領域的創新，創始團隊成員曾就職於 Yahoo、Twitter、Splunk、EMC 等知名大公司。

引用連結

[1] CDF: http://docs.cloudera.com/cdf-datahub/7.2.14/release-notes/topics/cdf-datahub-supported-partner-components.html
[2] Apache NiFi: http://nifi.apache.org/
[3] 文件: http://docs.cloudera.com/cdf-datahub/7.2.14/release-notes/topics/cdf-datahub-supported-partner-components.html
[4] Maven 中央程式碼倉庫: http://search.maven.org/search?q=g:io.streamnative.connectors%20nifi
[5] 構建: http://github.com/streamnative/pulsar-nifi-bundle
[6] 邊緣 AI 場景中 FLiPN 技術棧(Flink, NiFi, Pulsar)的應用: http://www.bilibili.com/video/BV1TP4y1P7TV
[7] 演示程式碼: http://github.com/tspannhw/FLiPN-NFT
[8] Producing and Consuming Pulsar messages with Apache NiFi: http://www.datainmotion.dev/2021/11/producing-and-consuming-pulsar-messages.html
[9] FLiP-Transit GitHub 倉庫: http://github.com/tspannhw/FLiP-Transit
[10] FLiPN-Demos GitHub 倉庫: http://github.com/tspannhw/FLiPN-Demos

---

▼ 關注「Apache Pulsar」，獲取更多技術幹貨 ▼

👇🏻加入 Apache Pulsar 中文交流群👇🏻