PySpark進階實戰：從MQTT感測資料到Spark Structured Streaming即時水質分析

從物聯網到大數據系列・第四篇

前面我們已學會使用Pandas整理智慧養殖資料，也用PySpark進行批次統計。但養殖池的水溫、pH與溶氧不會等到每天晚上才變化：感測器可能每幾秒就送出一筆數據。如何讓系統持續接收、分析並產生預警？這就是本篇的學習目標。

本篇以模擬養殖池資料示範 MQTT → Python橋接器 → JSON檔案 → Spark Structured Streaming → 統計與警示。先在一台電腦完成，再逐步替換成ESP32等真實設備。

一、先理解：MQTT與Spark負責不同的工作

技術	主要任務	智慧養殖範例
MQTT	發布與訂閱訊息	設備發布最新水質數據
Python橋接器	接收、驗證並持久化資料	把每筆MQTT訊息寫成獨立JSON檔
Spark Structured Streaming	持續處理新增資料	依養殖池統計平均水溫、溶氧
Dashboard／告警服務	呈現與通知	顯示趨勢、通知巡池人員

重要：標準Apache Spark不提供內建通用MQTT來源；不可直接假設 readStream.format("mqtt") 可以運作。本例採用檔案來源作為教學橋接。Structured Streaming通常以微批次執行，因此「持續處理」不等於毫秒級即時控制。

二、整體架構

模擬感測器 / ESP32
        |
        v
  MQTT Broker (1883)
        |
        v
 Python MQTT Subscriber
        |
        v
 data/inbox/*.json (每筆一個檔案)
        |
        v
 Spark readStream (JSON + Schema)
        |
        +--> 水質異常判斷
        +--> 依池別統計
        +--> Console / 後續資料庫

為方便理解，本篇的Broker以本機Mosquitto為例；若在區域網路中使用，請改成實際主機位址並設定帳號、TLS及網路存取限制。1883明文連線僅適合隔離的教學環境。

三、準備環境

建議使用Python虛擬環境，並依安裝的PySpark版本確認相容的Java版本。以下以終端機操作：

python -m venv .venv
# Windows PowerShell:
.venv\Scripts\Activate.ps1
# macOS / Linux:
# source .venv/bin/activate

python -m pip install pyspark paho-mqtt
python -c "import pyspark; print(pyspark.__version__)"

另外安裝並啟動Mosquitto MQTT Broker，確認本機 localhost:1883 可連線。以下三支程式請在同一個專案資料夾執行。若使用Windows，路徑請保持一致。

四、第一支程式：模擬感測器發布資料

建立 publisher.py。每兩秒發布一筆JSON，輪流模擬A、B兩個養殖池。數值與門檻均為教學假設，不代表特定魚種或養殖場的標準。

import json
import random
import time
from datetime import datetime, timezone
import paho.mqtt.client as mqtt

BROKER = "localhost"
TOPIC = "aquaculture/water"

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.connect(BROKER, 1883, 60)
client.loop_start()

try:
    count = 0
    while True:
        pond = "A" if count % 2 == 0 else "B"
        record = {
            "pond": pond,
            "event_time": datetime.now(timezone.utc).isoformat(),
            "temperature": round(random.uniform(26, 32), 2),
            "ph": round(random.uniform(6.5, 8.5), 2),
            "oxygen": round(random.uniform(3.5, 7.0), 2)
        }
        info = client.publish(TOPIC, json.dumps(record), qos=1)
        info.wait_for_publish()
        print("已發布:", record)
        count += 1
        time.sleep(2)
except KeyboardInterrupt:
    print("停止模擬")
finally:
    client.loop_stop()
    client.disconnect()

五、第二支程式：MQTT訂閱與JSON檔案橋接

建立 mqtt_bridge.py。Spark檔案串流會讀取「新出現的檔案」，因此不能把所有訊息一直附加到同一個檔案；本例每收到一筆有效資料，就先寫入暫存檔，再以原子重新命名移入監看資料夾。

import json
import os
import uuid
from pathlib import Path
import paho.mqtt.client as mqtt

BROKER = "localhost"
TOPIC = "aquaculture/water"
INBOX = Path("data/inbox")
TEMP = Path("data/tmp")
INBOX.mkdir(parents=True, exist_ok=True)
TEMP.mkdir(parents=True, exist_ok=True)

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        client.subscribe(TOPIC, qos=1)
        print("已訂閱:", TOPIC)
    else:
        print("連線失敗:", reason_code)

def on_message(client, userdata, msg):
    try:
        record = json.loads(msg.payload.decode("utf-8"))
        required = ("pond", "event_time", "temperature", "ph", "oxygen")
        if not isinstance(record, dict) or any(k not in record for k in required):
            raise ValueError("缺少必要欄位")
        if not isinstance(record["pond"], str) or not record["pond"]:
            raise ValueError("池別不正確")
        for k in ("temperature", "ph", "oxygen"):
            if isinstance(record[k], bool) or not isinstance(record[k], (int, float)):
                raise ValueError(f"{k}不是數值")
        filename = uuid.uuid4().hex + ".json"
        temp_path = TEMP / filename
        final_path = INBOX / filename
        with temp_path.open("w", encoding="utf-8") as f:
            json.dump(record, f, ensure_ascii=False)
            f.write("\n")
        os.replace(temp_path, final_path)
        print("已寫入:", final_path)
    except (UnicodeDecodeError, ValueError, TypeError) as e:
        print("略過無效訊息:", e)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, 60)
print("等待感測資料，Ctrl+C結束")
try:
    client.loop_forever()
except KeyboardInterrupt:
    client.disconnect()

MQTT QoS 1可能重送訊息。本例以唯一檔名避免覆寫，但不保證資料去重。正式系統應加入裝置ID、訊息ID、事件時間與去重機制；大量訊息也不適合長期採用「一筆一檔」。

六、第三支程式：PySpark Structured Streaming

建立 streaming.py。使用明確的Schema讀取新增JSON檔案，檢查基本數值範圍，再以 foreachBatch 對每個微批次做分組統計與異常判斷。

from pathlib import Path
from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, avg, count, min as spark_min, max as spark_max,
    when, sum as spark_sum
)
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType
)

Path("data/inbox").mkdir(parents=True, exist_ok=True)
Path("data/checkpoint").mkdir(parents=True, exist_ok=True)

spark = (SparkSession.builder
    .appName("AquacultureStreaming")
    .master("local[*]")
    .getOrCreate())
spark.sparkContext.setLogLevel("WARN")

schema = StructType([
    StructField("pond", StringType(), True),
    StructField("event_time", StringType(), True),
    StructField("temperature", DoubleType(), True),
    StructField("ph", DoubleType(), True),
    StructField("oxygen", DoubleType(), True),
])

raw = (spark.readStream
    .schema(schema)
    .option("maxFilesPerTrigger", 50)
    .json("data/inbox"))

clean = raw.filter(
    col("pond").isNotNull() &
    col("event_time").isNotNull() &
    col("temperature").between(0, 50) &
    col("ph").between(0, 14) &
    col("oxygen").between(0, 20)
)

# 以下門檻僅作教學示範，實際應依魚種、場域及專家建議設定
flagged = clean.withColumn(
    "warning",
    when((col("oxygen") < 5.0) | (col("temperature") > 30.0), 1)
    .otherwise(0)
)

def analyze_batch(batch_df, batch_id):
    if batch_df.isEmpty():
        return
    print(f"\n=== 微批次 {batch_id} ===")
    summary = batch_df.groupBy("pond").agg(
        count("*").alias("records"),
        avg("temperature").alias("avg_temp"),
        avg("oxygen").alias("avg_oxygen"),
        spark_min("ph").alias("min_ph"),
        spark_max("ph").alias("max_ph"),
        spark_sum("warning").alias("warning_count")
    )
    summary.show(truncate=False)
    batch_df.filter(col("warning") == 1).select(
        "pond", "event_time", "temperature", "oxygen"
    ).show(truncate=False)

query = (flagged.writeStream
    .foreachBatch(analyze_batch)
    .option("checkpointLocation", "data/checkpoint")
    .trigger(processingTime="5 seconds")
    .start())

try:
    query.awaitTermination()
except KeyboardInterrupt:
    query.stop()
finally:
    spark.stop()

這裡的 avg_temp、avg_oxygen 是當次微批次的平均值，並非全部歷史資料的累積平均。若要分析每5分鐘的事件時間視窗，下一階段可加入時間戳轉換、window()、withWatermark()與適當輸出模式。

七、實際執行順序

1. 啟動Mosquitto Broker。
2. 開啟終端機A：執行 python streaming.py。
3. 開啟終端機B：執行 python mqtt_bridge.py。
4. 開啟終端機C：執行 python publisher.py。
5. 觀察每5秒左右輸出的微批次統計；結束時以Ctrl+C停止。

第一次啟動Spark可能需要一些時間。若要重新完整測試，請先停止程式，再清除測試用的 data/inbox、data/tmp與 data/checkpoint，避免舊資料與檢查點影響結果。請勿在正式系統任意刪除checkpoint。

八、如何看懂結果？

假設某微批次中A池有三筆資料，溫度為28、29、31°C，溶氧為6.2、5.4、4.6 mg/L，則平均溫度為29.33°C、平均溶氧為5.40 mg/L；其中31°C及4.6 mg/L的紀錄會觸發本例示範門檻。這只是說明計算方式，不是實際場域監測結果。

九、從教學範例走向真實AIoT系統

教學版	正式場域需補強
本機MQTT Broker	帳號權限、TLS、Broker高可用性
每筆一個JSON檔	Kafka等訊息平台或支援的持續資料來源
單機local[*]	依資料量評估叢集與資源配置
固定示範門檻	魚種別規則、感測器校正與專家驗證
Console輸出	資料庫、Dashboard、告警通知
微批次內統計	事件時間視窗、延遲資料處理、去重與長期趨勢

若要真正實現大數據分析，可以讓多座養殖池長期累積資料，建立分區儲存與歷史分析；若要進一步做AIoT，則可加入異常偵測模型與人工覆核機制。溶氧控制等關鍵設備不應只依賴雲端分析結果進行安全控制，仍需保留在地控制、警報與人工介入機制。

十、學生挑戰題

1. 在JSON加入 device_id、message_id，思考如何避免重送造成重複統計。
2. 加入不同養殖池的自訂警示門檻。
3. 把微批次統計寫入資料庫，再由Django呈現。
4. 研究Spark事件時間視窗：當資料延遲到達時，統計結果如何改變？
5. 將模擬發布者替換成ESP32或Pico W，測試斷線與重連情境。

十一、延伸閱讀：智慧生活科技專業社群

• micro:bit和電腦的串列通訊程式加上MQTT
• 利用MQTT整合兩塊micro:bit來做燈光控制
• Django 6.0整合Django和MQTT
• Python asyncio TCP相關教學
• Apache Spark官方文件：Structured Streaming
• Eclipse Paho MQTT Python官方文件

十二、結語：讓感測資料成為可行動的知識

IoT讓我們取得資料，MQTT讓裝置交換訊息，PySpark讓資料持續被整理與分析，而AIoT則將分析結果轉化成輔助決策。學習的關鍵不是一次掌握所有平台，而是能清楚理解每一層負責什麼、如何驗證資料，以及何時需要更可靠的系統架構。

下一篇可以繼續探討：《Spark事件時間視窗實戰：用Watermark處理智慧養殖延遲感測資料》。