Apache Spark與PySpark入門：當智慧養殖感測資料變成大數據，Python該如何處理？

一座智慧養殖池每分鐘記錄水溫、pH與溶氧，資料量看起來不大；若擴展至多座養殖池、數百個感測器，持續記錄多年，資料就可能成為管理上的挑戰。上一篇我們利用Pandas清理與分析感測資料，這次要認識Apache Spark及其Python介面PySpark，了解何時需要分散式資料處理。

本篇學習目標：理解Pandas與Spark的差異、建立本機PySpark環境、使用DataFrame清理與彙整智慧養殖資料，並認識Spark與IoT資料平台的整合方式。

一、從一座養殖池開始：資料真的有那麼大嗎？

假設有100座養殖池，每池10個感測器，每分鐘記錄一次，每筆資料約100位元組，則一年約有：

100 池 × 10 感測器 × 60 分鐘 × 24 小時 × 365 天
= 525,600,000 筆紀錄／年
約 52.56 GB／年（以每筆100位元組粗估）

這只是示意估算，實際容量取決於資料格式、索引、壓縮、備份與紀錄方式。資料量大不代表一定要用Spark；如果資料庫查詢或Pandas仍可有效完成工作，通常沒有必要增加分散式系統的複雜度。

二、Pandas與Apache Spark有什麼不同？

比較項目	Pandas	Apache Spark／PySpark
主要用途	單機資料處理與探索	大規模資料處理與分散式運算
執行方式	主要在單一Python程序	可在單機或多節點叢集執行
記憶體需求	資料與操作常受單機記憶體限制	可將工作分割到多個執行器
學習門檻	較低	需要理解SparkSession、轉換與動作
適合案例	一週水質CSV、探索分析	多年跨場域大量紀錄、批次分析

PySpark是Apache Spark的Python API。它讓我們使用Python操作Spark DataFrame，但底層仍由Spark執行引擎規劃與執行工作。

三、先認識Spark的核心觀念

SparkSession：PySpark應用程式的主要入口。
DataFrame：有欄位名稱與資料型別的分散式表格。
Transformation（轉換）：例如filter()、select()、groupBy()，通常採延遲執行。
Action（動作）：例如show()、count()、collect()，會觸發計算。
Partition（分割區）：Spark平行處理資料的基本單位之一。

重要提醒：collect()會將結果帶回Driver程序。對大量資料應先彙總或限制筆數，避免記憶體不足。

四、安裝PySpark：從本機模式開始

建議使用支援版本的Python與Java環境，並依照Apache Spark官方安裝文件確認版本需求。以下使用虛擬環境及pip示範；實際安裝前請確認作業系統與Java相容性。

python -m venv .venv
# Windows PowerShell：.venv\Scripts\Activate.ps1
# macOS / Linux：source .venv/bin/activate
python -m pip install pyspark

新增檔案test_spark.py：

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .master("local[*]")
         .appName("AquacultureIntro")
         .getOrCreate())

print("Spark version:", spark.version)
spark.range(5).show()
spark.stop()

在命令列執行python test_spark.py。local[*]表示使用本機可用的邏輯核心執行，不代表已建立多台電腦的叢集。

五、智慧養殖PySpark實作：建立感測資料

以下使用模擬資料，不是水井村實測數據。欄位包括養殖池、時間、水溫（°C）、pH及溶氧（mg/L）。其中一筆溶氧缺漏，另一筆pH超出合理資料範圍，供清理練習。

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType
)

spark = (SparkSession.builder
         .master("local[*]")
         .appName("PondWaterQuality")
         .getOrCreate())

rows = [
    ("A", "2026-11-01 08:00:00", 27.5, 7.8, 6.2),
    ("A", "2026-11-01 09:00:00", 28.0, 7.7, 5.8),
    ("A", "2026-11-01 10:00:00", 28.4, 7.9, None),
    ("B", "2026-11-01 08:00:00", 29.1, 7.6, 4.9),
    ("B", "2026-11-01 09:00:00", 28.7, 15.0, 5.1),
    ("B", "2026-11-01 10:00:00", 28.3, 7.5, 5.4),
]
schema = StructType([
    StructField("pond", StringType(), False),
    StructField("time_text", StringType(), False),
    StructField("temperature", DoubleType(), True),
    StructField("ph", DoubleType(), True),
    StructField("oxygen", DoubleType(), True),
])

df = spark.createDataFrame(rows, schema)
df = df.withColumn("timestamp", F.to_timestamp("time_text"))
df.select("pond", "timestamp", "temperature", "ph", "oxygen").show()

六、清理資料：缺漏值與異常值

資料清理必須區分不可能的數值與可能需要警報的真實異常。例如pH=15超出一般水溶液常見量測範圍，適合標記為資料品質問題；但偏低的溶氧可能正是需要保留的危險訊號，不能為了讓統計好看而刪掉。

# 將超出示範資料品質界限的pH設為缺漏
clean = df.withColumn(
    "ph_clean",
    F.when(F.col("ph").between(0.0, 14.0), F.col("ph"))
     .otherwise(F.lit(None).cast("double"))
)

# 示範策略：各養殖池的平均值補缺漏
means = clean.groupBy("pond").agg(
    F.avg("oxygen").alias("mean_oxygen"),
    F.avg("ph_clean").alias("mean_ph")
)
clean = (clean.join(means, on="pond", how="left")
    .withColumn("oxygen_final", F.coalesce("oxygen", "mean_oxygen"))
    .withColumn("ph_final", F.coalesce("ph_clean", "mean_ph")))

clean.select("pond", "timestamp", "oxygen_final", "ph_final").show()

教學上使用平均值補缺漏便於觀察，但真實監控系統應另外保留原始值、缺漏旗標、補值方法及資料品質紀錄；若要進行安全警報判斷，不能直接把補值當成真實量測。

七、用groupBy進行各池統計

summary = clean.groupBy("pond").agg(
    F.round(F.avg("temperature"), 2).alias("avg_temp"),
    F.round(F.min("temperature"), 2).alias("min_temp"),
    F.round(F.max("temperature"), 2).alias("max_temp"),
    F.round(F.avg("oxygen_final"), 2).alias("avg_oxygen"),
    F.round(F.avg("ph_final"), 2).alias("avg_ph"),
    F.count("*").alias("records")
)
summary.orderBy("pond").show()

預期每座池各有3筆紀錄；A池平均水溫約27.97°C，B池約28.70°C。這些都是根據模擬資料算出的教學結果，不代表實際養殖環境的適宜標準。

八、Spark也能做視覺化嗎？

Spark適合先進行大規模清理與彙總，再將足夠小的結果轉成Pandas，交給Matplotlib繪圖。不要把數千萬筆原始資料直接轉成Pandas。

import matplotlib.pyplot as plt

# 只有兩個池的彙總結果，安全地轉為Pandas
plot_df = summary.orderBy("pond").toPandas()
plt.bar(plot_df["pond"], plot_df["avg_oxygen"])
plt.xlabel("Pond")
plt.ylabel("Average dissolved oxygen (mg/L)")
plt.title("Average dissolved oxygen by pond (simulated)")
plt.tight_layout()
plt.savefig("pond_oxygen_summary.png", dpi=150)
plt.show()

本範例使用英文圖表標籤，以避免不同電腦缺少中文字型時出現方框。

九、資料儲存：從CSV到Parquet

CSV適合入門及跨工具交換；當資料量增長時，Parquet這類欄式儲存格式通常更適合分析型工作負載。

# Spark寫出時通常會建立資料夾及part檔案
summary.write.mode("overwrite").parquet("output/pond_summary_parquet")

# 若需CSV，仍會輸出成資料夾
summary.write.mode("overwrite").option("header", True).csv(
    "output/pond_summary_csv"
)

spark.stop()

請注意：Spark的write.csv()不是直接寫出單一CSV檔，而是輸出包含part檔案的目錄。大型資料集也不建議為了合併成單檔而隨意使用coalesce(1)。

十、從IoT走向大數據與AIoT：完整架構

感測器（ESP32／Pico W）→ MQTT或HTTP → 資料接收服務 → 資料庫／檔案儲存 → Spark批次分析 → Dashboard／AI模型 → 人員決策。

例如ESP32將水溫、pH與溶氧傳至伺服器，Django提供查詢介面；每天或每小時由Spark分析大量歷史資料，再提供趨勢圖與異常報表。若未來需要持續處理即時資料流，可再研究Spark Structured Streaming與訊息系統的整合。本篇範例只有批次分析，尚未實作MQTT接收或即時串流。

十一、學生練習與挑戰題

將每池資料擴充到24小時，每小時一筆，加入一筆重複紀錄，練習dropDuplicates()。
比較「刪除缺漏資料」與「各池平均值補值」對平均溶氧的影響。
增加日期欄位，使用to_date()與groupBy()製作每日報表。
設計資料品質欄位，例如is_missing與is_out_of_range，保留原始感測值。
討論何時Pandas已經足夠、何時才值得改用Spark，說明記憶體、資料量與維運成本。

十二、延伸閱讀：智慧生活科技專業社群系列文章

micro:bit和電腦的串列通訊程式加上MQTT：認識感測資料如何傳送。
利用MQTT整合兩塊micro:bit來做燈光控制：理解發布與訂閱。
Django 6.0整合Django和MQTT：從裝置通訊走向網站。
讓ChatGPT產生樞紐分析表來分析漁貨交易資料：從Pandas開始學資料分析。
HTTP與REST API｜Pico 2 W：認識裝置與網路服務的資料交換。

官方參考：PySpark Getting Started、Apache Spark Documentation。

結語：大數據不是從Spark開始，而是從正確理解資料開始

IoT讓我們取得資料，Pandas幫助我們理解資料，Spark讓我們在資料規模增長時仍能有效分析，而AIoT則是將資料分析與智慧決策整合到真實場域。對學生而言，最重要的不是立即部署複雜叢集，而是先掌握資料品質、分析問題與工具選擇，再逐步從單機走向分散式運算。

智慧生活科技專業社群

2026年11月11日星期三