資料串流導論

使用 Python 分析 IoT 資料

Matthias Voppichler

IT Developer

什麼是資料串流

持續不斷的資料串流
範例
- Twitter 訊息
- 線上新聞文章
- 影像串流
- 感測器資料（IoT）
- 市場委託單（金融）

什麼是資料串流

持續不斷的資料串流
範例
- Twitter 訊息
- 線上新聞文章
- 影像串流
- 感測器資料（IoT）
- 市場委託單（金融）

MQTT

訊息通訊協定
發佈／訂閱
小型占用

伺服器 -> 作為訊息 Broker

用戶端：

連線至 Broker
發佈資料
訂閱主題

Message Queuing Telemetry Transport

MQTT 資料流—從生產者 -> 到 Broker -> 到消費者

Python 函式庫

Eclipse Paho™ MQTT Python Client

# Import MQTT library
import paho.mqtt

更多資訊與文件請見 GitHub https://github.com/eclipse/paho.mqtt.python

單一訊息

import paho.mqtt.subscribe as subscribe
msg = subscribe.simple("paho/test/simple", 
                       hostname="test.mosquitto.org")

print(f"{msg.topic}, {msg.payload}")

輸出：

paho/test/simple, {"time": 1549481572, "humidity": 77, "temp": 21}

回呼函式

def on_message(client, userdata, message):

    print(f"{message.topic} : {message.payload}")

參數

client - 用戶端實例
userdata - 私有使用者資料
message - MQTTMessage 實例

回呼函式

import paho.mqtt.subscribe as subscribe


subscribe.callback(on_message, 
                   topics="datacamp/roomtemp", 
                   hostname="test.mosquitto.org")

MQTT 訂閱

import paho.mqtt.subscribe as subscribe

def on_message(client, userdata, message):
    print("{} : {}".format(message.topic, message.payload))

subscribe.callback(on_message, 
                   topics="datacamp/roomtemp", 
                   hostname="test.mosquitto.org")

datacamp/roomtemp : b'{"time": 1543344857, "hum": 34, "temp": 24}'
datacamp/roomtemp : b'{"time": 1543344858, "hum": 35, "temp": 23}'
datacamp/roomtemp : b'{"time": 1543344860, "hum": 36, "temp": 22}'
datacamp/roomtemp : b'{"time": 1543344946, "hum": 37, "temp": 22}'
datacamp/roomtemp : b'{"time": 1543345010, "hum": 36, "temp": 13}'

一起來練習吧！

使用 Python 分析 IoT 資料