NiFi 與 Kafka 資料管道建構與最佳實務

– 14 分鐘閱讀

在資料工程領域中，建構穩固且高效的資料管道至關重要。本文將深入探討如何運用 Apache NiFi 和 Kafka 建構批次和流式資料處理管道，並涵蓋最佳實務、效能調校、以及與 Python 的整合應用。我們將探討 NiFi Registry 的版本控制和 git-persistence 功能，並說明如何透過 NiFi 的 GUI 和日誌進行管道監控。此外，文章也將闡述如何使用 NiFi 的 REST API 與 Python 程式碼互動，實作自動化監控和管理。更進一步，我們將探討如何建構測試和生產環境、設定資料函式庫和資料湖，並逐步建構完整的生產資料管線，同時也將探討背壓處理和處理器群組最佳化等議題。最後，文章將介紹 Kafka 的佈署、主題建立、生產者和消費者設定，以及如何結合 Kafka 和 NiFi 構建實時流式資料管道，並比較流式處理和批次處理的差異。

使用NiFi Registry

組態完成後，可以使用NiFi Registry來管理資料管道。以下是使用NiFi Registry的步驟：

  1. 新增NiFi Registry到NiFi：需要將NiFi Registry新增到NiFi中，然後可以使用NiFi Registry來管理資料管道。
  2. 版本控制資料管道：可以使用NiFi Registry來版本控制資料管道，包括建立、更新和刪除資料管道。
  3. 使用git-persistence：NiFi Registry支援git-persistence，允許使用Git來儲存和管理資料管道的版本。

監控資料管道

除了版本控制，監控資料管道也是重要的。Apache NiFi提供了一個GUI介面，允許使用者監控資料管道的執行狀態。以下是監控資料管道的步驟：

  1. 使用GUI介面：可以使用NiFi的GUI介面來監控資料管道的執行狀態，包括資料管道的輸入、輸出和錯誤資訊。
  2. 檢視資料管道的執行日誌：可以檢視資料管道的執行日誌，包括資料管道的執行時間、輸入資料量和輸出資料量等。
# NiFi Registry的基本組態
nifi_registry_url = "http://localhost:18080/nifi-registry"
nifi_registry_username = "admin"
nifi_registry_password = "admin"

# 新增NiFi Registry到NiFi
nifi_url = "http://localhost:8080/nifi"
nifi_username = "admin"
nifi_password = "admin"

# 版本控制資料管道
pipeline_name = "my_pipeline"
pipeline_version = "1.0"

# 使用git-persistence
git_url = "https://github.com/myrepo/myrepo.git"
git_username = "myusername"
git_password = "mypassword"

圖表翻譯：

  flowchart TD
    A[安裝NiFi Registry] --> B[組態NiFi Registry]
    B --> C[新增NiFi Registry到NiFi]
    C --> D[版本控制資料管道]
    D --> E[使用git-persistence]
    E --> F[監控資料管道]

以上是使用NiFi Registry來管理資料管道的基本步驟，包括安裝、組態、版本控制和監控等。

監控NiFi的效能

NiFi是一個強大的資料處理工具，能夠高效地處理大量的資料。但是，為了確保NiFi的效能和穩定性，監控其狀態和效能是非常重要的。

監控NiFi的狀態列

NiFi提供了一個狀態列，允許使用者監控NiFi的實時狀態，包括處理器的執行狀態、記憶體使用情況等。這個狀態列可以幫助使用者快速地發現NiFi的問題，並進行相應的調整和最佳化。

監控NiFi的處理器

NiFi的處理器是其核心元件，負責處理和轉換資料。監控處理器的執行狀態和效能，可以幫助使用者瞭解NiFi的整體效能和瓶頸。NiFi提供了多種監控處理器的方法，包括使用REST API、使用圖形化介面等。

使用Python與NiFi REST API

NiFi提供了一個REST API，允許使用者使用Python等程式語言來監控和控制NiFi。這個API提供了多種功能，包括監控NiFi的狀態、查詢處理器的執行狀態等。使用Python與NiFi REST API，可以幫助使用者自動化NiFi的監控和維護。

佈署資料管道

佈署資料管道是NiFi的最終目標。為了確保資料管道的穩定性和效能，需要進行多種最佳化和調整。

完成資料管道的佈署

完成資料管道的佈署需要考慮多種因素，包括資料來源、資料處理、資料儲存等。NiFi提供了多種工具和功能，幫助使用者完成資料管道的佈署。

背壓

背壓是NiFi的一個重要概念，指的是當NiFi的處理器無法處理資料的速度時，會產生的壓力。背壓可以導致NiFi的效能下降和資料丟失。為了避免背壓，需要進行多種最佳化和調整，包括增加處理器的執行緒、最佳化資料處理等。

改進處理器群組

處理器群組是NiFi的一個重要功能，允許使用者將多個處理器組合成一個群組。改進處理器群組可以幫助使用者提高NiFi的效能和穩定性。

使用NiFi變數登記

NiFi變數登記是NiFi的一個重要功能，允許使用者定義和管理變數。使用NiFi變數登記可以幫助使用者簡化資料處理和最佳化NiFi的效能。

佈署資料管道

佈署資料管道需要考慮多種因素，包括資料來源、資料處理、資料儲存等。NiFi提供了多種工具和功能，幫助使用者佈署資料管道。

使用最簡單的策略

使用最簡單的策略可以幫助使用者快速地佈署資料管道。但是，這個策略可能不能提供最佳的效能和穩定性。

使用中間策略

使用中間策略可以幫助使用者在效能和穩定性之間找到一個平衡點。這個策略需要考慮多種因素，包括資料來源、資料處理、資料儲存等。

# 監控NiFi的狀態列
import requests

nifi_url = "http://localhost:8080/nifi-api"
response = requests.get(nifi_url + "/process-groups/root")

# 監控NiFi的處理器
import requests

nifi_url = "http://localhost:8080/nifi-api"
response = requests.get(nifi_url + "/processors")

# 使用Python與NiFi REST API
import requests

nifi_url = "http://localhost:8080/nifi-api"
response = requests.get(nifi_url + "/process-groups/root")

內容解密：

以上程式碼示範瞭如何使用Python與NiFi REST API來監控NiFi的狀態列和處理器。這些程式碼可以幫助使用者自動化NiFi的監控和維護。

圖表翻譯：

  graph LR
    A[監控NiFi的狀態列] --> B[監控NiFi的處理器]
    B --> C[使用Python與NiFi REST API]
    C --> D[佈署資料管道]
    D --> E[完成資料管道的佈署]
    E --> F[背壓]
    F --> G[改進處理器群組]
    G --> H[使用NiFi變數登記]
    H --> I[佈署資料管道]
    I --> J[使用最簡單的策略]
    J --> K[使用中間策略]

此圖表示範了NiFi的監控和佈署流程。這個流程包括監控NiFi的狀態列、監控NiFi的處理器、使用Python與NiFi REST API、佈署資料管道等步驟。

建立生產資料管線

在實際應用中，建立一個穩定且高效的生產資料管線是非常重要的。這個過程涉及到多個步驟，包括建立測試和生產環境、建立資料函式庫、填充資料湖、建造生產資料管線等。

建立測試和生產環境

為了確保資料管線的穩定性和可靠性，需要建立測試和生產環境。這兩個環境應該是隔離的，以避免測試過程中出現的錯誤影響到生產環境。測試環境可以用於測試和驗證資料管線的功能，生產環境則用於實際的資料處理和分析。

建立資料函式庫

建立資料函式庫是資料管線的基礎。需要根據業務需求建立相應的資料函式庫，例如建立一個用於儲存原始資料的資料函式庫，或者建立一個用於儲存處理後資料的資料函式庫。

填充資料湖

資料湖是用於儲存原始資料的倉函式庫。填充資料湖的過程涉及到從各個來源收集資料，然後將其儲存到資料湖中。這個過程可以使用各種工具和技術，例如使用 Apache NiFi 或 Apache Kafka 來收集和處理資料。

建造生產資料管線

建造生產資料管線的目的是將資料從資料湖中提取出來，然後進行處理和分析。這個過程涉及到多個步驟，包括讀取資料湖、掃描資料湖、插入資料到臨時表、查詢臨時表、驗證臨時資料等。

讀取資料湖

讀取資料湖的過程涉及到從資料湖中提取出所需的資料。這個過程可以使用各種工具和技術，例如使用 Apache Hive 或 Apache Spark 來讀取資料。

掃描資料湖

掃描資料湖的過程涉及到掃描資料湖中的所有資料，然後將其儲存到臨時表中。這個過程可以使用各種工具和技術，例如使用 Apache Hive 或 Apache Spark 來掃描資料。

插入資料到臨時表

插入資料到臨時表的過程涉及到將掃描出的資料插入到臨時表中。這個過程可以使用各種工具和技術，例如使用 Apache Hive 或 Apache Spark 來插入資料。

查詢臨時表

查詢臨時表的過程涉及到查詢臨時表中的資料，然後進行分析和處理。這個過程可以使用各種工具和技術，例如使用 Apache Hive 或 Apache Spark 來查詢資料。

驗證臨時資料

驗證臨時資料的過程涉及到驗證臨時表中的資料，然後確保其正確性和完整性。這個過程可以使用各種工具和技術，例如使用 Apache Hive 或 Apache Spark 來驗證資料。

插入倉函式庫

插入倉函式庫的過程涉及到將驗證後的資料插入到倉函式庫中。這個過程可以使用各種工具和技術，例如使用 Apache Hive 或 Apache Spark 來插入資料。

# 讀取資料湖
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Read Data Lake").getOrCreate()
data_lake_df = spark.read.format("parquet").load("data_lake_path")

# 掃描資料湖
data_lake_df.createOrReplaceTempView("data_lake_temp_view")

# 插入資料到臨時表
temp_table_df = spark.sql("SELECT * FROM data_lake_temp_view")
temp_table_df.write.format("parquet").save("temp_table_path")

# 查詢臨時表
temp_table_df = spark.read.format("parquet").load("temp_table_path")
temp_table_df.show()

# 驗證臨時資料
from pyspark.sql.functions import col
validated_df = temp_table_df.filter(col("column_name").isNotNull())

# 插入倉函式庫
validated_df.write.format("parquet").save("warehouse_path")

圖表翻譯：

此圖示為建立生產資料管線的流程圖，描述了從讀取資料湖到插入倉函式庫的各個步驟。

  flowchart TD
    A[讀取資料湖] --> B[掃描資料湖]
    B --> C[插入資料到臨時表]
    C --> D[查詢臨時表]
    D --> E[驗證臨時資料]
    E --> F[插入倉函式庫]

建立實時資料管道：Kafka 叢集佈署

在前面的章節中，我們討論了批次處理資料管道的建立。然而，在現代的大資料應用中，實時資料處理越來越重要。為了滿足這種需求，我們需要建立實時資料管道。Apache Kafka 是一個流行的分散式流式處理平臺，非常適合建立實時資料管道。

建立 Kafka 叢集

要建立一個 Kafka 叢集，首先需要建立 ZooKeeper 和 Kafka 叢集。ZooKeeper 是一個分散式協調服務，負責管理 Kafka 叢集的後設資料。

下載 Kafka 和設定環境

首先，需要下載 Kafka 的二進位制包，並設定環境變數。假設我們下載了 Kafka 3.1.0 版本，則可以如下設定環境變數：

export KAFKA_HOME=/path/to/kafka_2.13-3.1.0
export PATH=$KAFKA_HOME/bin:$PATH

組態 ZooKeeper 和 Kafka

接下來，需要組態 ZooKeeper 和 Kafka。ZooKeeper 的組態檔案位於 $KAFKA_HOME/config/zookeeper.properties，而 Kafka 的組態檔案位於 $KAFKA_HOME/config/server.properties。我們需要設定 ZooKeeper 的連線埠和 Kafka 的 broker 埠。

# zookeeper.properties
clientPort=2181

# server.properties
broker.id=0
listeners=PLAINTEXT://:9092

啟動 ZooKeeper 和 Kafka 叢集

組態完成後，需要啟動 ZooKeeper 和 Kafka 叢集。可以使用以下命令啟動 ZooKeeper：

zkServer.sh start

然後，啟動 Kafka 叢集：

kafka-server-start.sh $KAFKA_HOME/config/server.properties

測試 Kafka 叢集

啟動 Kafka 叢集後，需要測試是否正常運作。可以使用 Kafka 的命令列工具建立一個主題，然後生產和消費訊息。

kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 test-topic
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning

這樣就完成了 Kafka 叢集的佈署和測試。接下來，我們可以使用 Kafka 建立實時資料管道，實作資料的實時處理和分析。

內容解密：

在上面的過程中，我們使用了 Kafka 的命令列工具建立了一個主題，然後生產和消費訊息。這個過程可以用以下的流程圖來描述：

  flowchart TD
    A[建立主題] --> B[生產訊息]
    B --> C[消費訊息]
    C --> D[驗證結果]

這個流程圖描述了 Kafka 叢集的基本運作過程。首先，建立一個主題，然後生產訊息，最後消費訊息並驗證結果。

圖表翻譯：

上面的流程圖描述了 Kafka 叢集的基本運作過程。這個過程可以分為四個步驟：建立主題、生產訊息、消費訊息和驗證結果。每個步驟都對應了一個特定的 Kafka 命令列工具。透過這個流程圖，可以清晰地看到 Kafka 叢集的運作過程和每個步驟的關係。

流式資料處理與 Apache Kafka

Apache Kafka 是一種流行的流式資料處理平臺，廣泛應用於大資料處理、物聯網、實時分析等領域。瞭解 Kafka 的基本概念和工作原理對於構建高效的流式資料管道至關重要。

日誌的理解

日誌（log）是 Kafka 中的一個基本概念，代表了一系列按時間順序排列的事件或記錄。日誌可以來自各種來源，如應用程式、伺服器、感應器等。Kafka 將這些日誌組織成一個稱為「主題」（topic）的集合，以便於管理和處理。

Kafka 如何使用日誌

Kafka 使用日誌來儲存和管理流式資料。當資料被寫入 Kafka 時，它會被分割成多個部分，並儲存於不同的Broker節點上。這種設計使得 Kafka 能夠提供高用性和高吞吐量的流式資料處理能力。

主題（Topic）

主題是 Kafka 中的一個基本概念，代表了一類相關的日誌或事件。主題可以被視為一個名稱空間，用於組織和管理相關的日誌。Kafka 支援建立多個主題，以便於管理和處理不同型別的流式資料。

Kafka 生產者和消費者

Kafka 生產者（producer）負責將資料寫入 Kafka 主題中。生產者可以是應用程式、伺服器、感應器等。Kafka 消費者（consumer）負責從 Kafka 主題中讀取資料。消費者可以是應用程式、伺服器、資料函式庫等。

使用 Kafka 和 NiFi 建立資料管道

Apache NiFi 是一個開源的資料整合平臺，用於管理和處理流式資料。Kafka 和 NiFi 可以整合使用，建立高效的流式資料管道。NiFi 可以用於收集、轉換和路由資料到 Kafka 主題中。

Kafka 生產者

Kafka 生產者負責將資料寫入 Kafka 主題中。生產者可以組態不同的引數，例如批次大小、壓縮等，以最佳化資料寫入的效率。

// Kafka 生產者範例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "Hello, World!"));

Kafka 消費者

Kafka 消費者負責從 Kafka 主題中讀取資料。消費者可以組態不同的引數，例如群組 ID、偏移量等，以控制資料的讀取。

// Kafka 消費者範例
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singleton("my-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record.value());
    }
}

流式處理與批次處理的區別

流式處理和批次處理是兩種不同的資料處理模式。流式處理是指實時處理資料，通常用於需要即時反應的應用程式。批次處理是指將資料分成批次，然後進行處理，通常用於需要批次處理的大量資料。Kafka 支援兩種模式的資料處理，讓開發者可以根據不同的需求選擇合適的處理模式。

實時邊緣資料處理技術

隨著物聯網（IoT）和邊緣計算的發展，實時資料處理的需求不斷增長。Apache Kafka、Apache Spark和MiNiFi是實時資料處理的三個重要工具。這篇文章將介紹如何使用Python和這些工具來實作實時邊緣資料處理。

Apache Kafka簡介

Apache Kafka是一個分散式流式處理平臺，能夠高效地處理大量的實時資料。Kafka的核心概念包括主題（Topic）、生產者（Producer）和消費者（Consumer）。生產者將資料釋出到主題中，消費者則從主題中訂閱和接收資料。

寫入Kafka的Python生產者

以下是使用Python寫入Kafka的簡單生產者示例：

from kafka import KafkaProducer

# 建立Kafka生產者
producer = KafkaProducer(bootstrap_servers='localhost:9092')

# 釋出訊息到主題
producer.send('my_topic', value='Hello, Kafka!'.encode('utf-8'))

讀取Kafka的Python消費者

以下是使用Python讀取Kafka的簡單消費者示例：

from kafka import KafkaConsumer

# 建立Kafka消費者
consumer = KafkaConsumer('my_topic', bootstrap_servers='localhost:9092')

# 讀取訊息從主題
for message in consumer:
    print(message.value.decode('utf-8'))

Apache Spark簡介

Apache Spark是一個統一的分析引擎，能夠高效地處理大量的資料。Spark的核心概念包括RDD（Resilient Distributed Dataset）和DataFrame。RDD是一個分散式的資料集合，DataFrame是一個有結構的資料集合。

安裝和執行Spark

要使用Spark，需要先安裝和執行Spark。以下是安裝和執行Spark的步驟：

# 下載Spark
wget https://apache.org/dist/spark/spark-3.2.1/spark-3.2.1-bin-hadoop2.7.tgz

# 解壓Spark
tar -xvf spark-3.2.1-bin-hadoop2.7.tgz

# 執行Spark
./spark-3.2.1-bin-hadoop2.7/bin/spark-shell

安裝和組態PySpark

PySpark是Spark的Python API。要使用PySpark，需要先安裝和組態PySpark。以下是安裝和組態PySpark的步驟：

# 安裝PySpark
pip install pyspark

# 組態PySpark
export PYSPARK_SUBMIT_ARGS="--master local[2] pyspark-shell"
export SPARK_HOME=/path/to/spark

處理資料以PySpark

以下是使用PySpark處理資料的簡單示例：

from pyspark.sql import SparkSession

# 建立SparkSession
spark = SparkSession.builder.appName("My App").getOrCreate()

# 建立DataFrame
data = [("John", 23), ("Mary", 31), ("David", 42)]
df = spark.createDataFrame(data, ["Name", "Age"])

# 顯示DataFrame
df.show()

MiNiFi簡介

MiNiFi是一個輕量級的資料收集和處理工具，能夠高效地收集和處理實時資料。MiNiFi的核心概念包括agent和流程。

安裝和執行MiNiFi

要使用MiNiFi，需要先安裝和執行MiNiFi。以下是安裝和執行MiNiFi的步驟：

# 下載MiNiFi
wget https://nifi.apache.org/minifi/downloads.html

# 解壓MiNiFi
tar -xvf minifi-0.7.0.tar.gz

# 執行MiNiFi
./minifi-0.7.0/bin/minifi.sh

結合Kafka、Spark和MiNiFi

以下是結合Kafka、Spark和MiNiFi的簡單示例：

from kafka import KafkaProducer
from pyspark.sql import SparkSession
from minifi import Minifi

# 建立Kafka生產者
producer = KafkaProducer(bootstrap_servers='localhost:9092')

# 建立SparkSession
spark = SparkSession.builder.appName("My App").getOrCreate()

# 建立MiNiFi agent
agent = Minifi()

# 收集資料
data = agent.collect()

# 釋出資料到Kafka
producer.send('my_topic', value=data)

# 讀取資料從Kafka
consumer = KafkaConsumer('my_topic', bootstrap_servers='localhost:9092')

# 處理資料以Spark
df = spark.createDataFrame(consumer, ["Name", "Age"])

# 顯示DataFrame
df.show()

資料工程入門

資料工程是資料科學和分析的基礎，對於所有企業來說都是非常重要的。這本章將引導您探索使用Python進行資料工程的各種工具和方法。

綜觀資料工程領域的技術發展趨勢，從批次處理到流式處理，再到邊緣計算，資料處理的效率和實時性要求越來越高。本文深入探討了從NiFi Registry的版本控制到Kafka及Spark的流式資料處理，以及MiNiFi在邊緣計算的應用，涵蓋了資料工程的關鍵環節。分析各技術的實作細節可以發現，資料管線的構建需要考量資料來源、處理方式、儲存目標以及不同技術間的整合。目前，邊緣計算的資料處理仍面臨挑戰，例如資源受限和網路延遲等問題。玄貓認為，未來資料工程的發展方向將更注重邊緣計算與雲端計算的協同，利用MiNiFi等輕量級工具在邊緣端進行預處理，再將關鍵資料傳輸至雲端進行深度分析，以實作更高效、更即時的資料洞察。隨著5G和物聯網的普及，邊緣智慧將成為主流，而精通Kafka、Spark和MiNiFi等技術的資料工程師將扮演更重要的角色。

玄貓

玄貓

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。