数据流简介

用 Python 分析 IoT 数据

Matthias Voppichler

IT Developer

什么是数据流

持续的数据流
示例
- 推文
- 在线新闻文章
- 视频流
- 传感器数据（IoT）
- 市场订单（金融）

什么是数据流

持续的数据流
示例
- 推文
- 在线新闻文章
- 视频流
- 传感器数据（IoT）
- 市场订单（金融）

MQTT

消息协议
发布/订阅
轻量级

服务器 -> 充当消息代理

客户端：

连接到代理
发布数据
订阅主题

Message Queuing Telemetry Transport

MQTT 数据流——从生产者 -> 到代理 -> 到消费者

Python 库

Eclipse Paho™ MQTT Python 客户端

# 导入 MQTT 库
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}'

Passons à la pratique !

用 Python 分析 IoT 数据