数据采集基础：传感器数据读取、日志文件采集、API数据拉取、数据格式解析
大家好，我是老张。在边缘计算这个领域摸爬滚打了快十年，今天咱们聊聊数据采集这件事。
很多人觉得数据采集嘛，不就是把数据拿过来？其实没那么简单。我在项目里吃过不少亏，比如传感器数据读着读着就丢了，日志文件解析到一半程序崩了，API拉取数据被限流……这些坑，今天咱们一个一个填上。
核心观点：数据采集是边缘计算的「第一公里」。这步没走好，后面清洗、预处理全是白搭。
一、传感器数据读取：从物理世界到数字世界
传感器是边缘设备最常用的数据源。温度、湿度、压力、振动……这些物理量怎么变成计算机能理解的数据？
我习惯把传感器读取分成三步：初始化、连续读取、异常处理。
1.1 常见传感器接口
| 接口类型 | 典型传感器 | 读取方式 | 注意事项 |
|---|---|---|---|
| I2C | 温湿度传感器（如SHT30） | 地址+寄存器读取 | 地址冲突、时钟延展 |
| SPI | 加速度计（如MPU6050） | 片选+时钟同步 | 线序、速率匹配 |
| UART | GPS模块、激光雷达 | 串口数据流解析 | 波特率、数据帧格式 |
| GPIO | 开关量、脉冲计数 | 电平检测/中断 | 去抖动、中断优先级 |
1.2 实战代码：读取I2C温湿度传感器
拿我常用的SHT30来说，代码其实不复杂。但有个坑——传感器刚上电时数据不稳定。我曾经因为这个，在产线上排查了整整两天。
import smbus
import time
# 初始化I2C总线
bus = smbus.SMBus(1)
# SHT30默认地址
SHT30_ADDR = 0x44
def read_sht30():
try:
# 发送测量命令
bus.write_i2c_block_data(SHT30_ADDR, 0x2C, [0x06])
time.sleep(0.015) # 等待测量完成
# 读取6字节数据
data = bus.read_i2c_block_data(SHT30_ADDR, 0x00, 6)
# 解析温度（高字节+低字节）
temp_raw = ((data[0] << 8) | data[1])
temperature = -45 + 175 * (temp_raw / 65535.0)
# 解析湿度
hum_raw = ((data[3] << 8) | data[4])
humidity = 100 * (hum_raw / 65535.0)
return round(temperature, 2), round(humidity, 2)
except Exception as e:
print(f"读取失败: {e}")
return None, None
# 连续读取
while True:
temp, hum = read_sht30()
if temp:
print(f"温度: {temp}°C, 湿度: {hum}%")
time.sleep(2) # 每2秒采集一次我的经验：传感器读取一定要加超时机制。有些传感器在恶劣环境下会「死掉」，不加超时整个程序就卡死了。我一般设500ms超时，读不到就重启传感器。
二、日志文件采集：从文本中提取价值
边缘设备上日志文件是最常见的数据来源。系统日志、应用日志、访问日志……说白了，日志就是设备在「自言自语」。我们要做的，就是听懂它在说什么。
2.1 日志采集的三种模式
- 全量采集：一次性读取整个文件。适合小文件，比如配置文件。
- 增量采集：只读取新增内容。适合持续运行的日志，比如syslog。
- 轮询采集：定时检查文件变化。适合日志文件会滚动的情况。
2.2 实战：用Python实现日志增量采集
我记得有一次，客户说他们的设备每天产生2GB日志，全量读取根本扛不住。后来我用了tail -f的思路，只读新增行。
import time
import os
def tail_log(file_path, callback):
"""模拟tail -f，只读取新增内容"""
# 先定位到文件末尾
with open(file_path, 'r') as f:
f.seek(0, os.SEEK_END)
while True:
line = f.readline()
if line:
# 回调函数处理每一行
callback(line.strip())
else:
time.sleep(0.1) # 没新内容就歇会儿
def process_line(line):
"""解析日志行，提取关键信息"""
# 假设日志格式: [2024-01-15 10:30:45] ERROR: 连接超时
if 'ERROR' in line:
print(f"[告警] {line}")
elif 'WARN' in line:
print(f"[注意] {line}")
# 开始采集
tail_log('/var/log/syslog', process_line)注意：日志文件可能被轮转（logrotate）。如果文件被重命名或删除，你的程序还在读旧文件句柄，那就读不到新内容了。我一般会监控文件inode变化，发现变了就重新打开文件。
三、API数据拉取：和云端打交道
边缘设备经常需要从云端拉取数据，比如配置信息、模型参数、业务数据。API调用看着简单，但坑不少。
3.1 API调用的关键点
- 认证方式：API Key、Token、OAuth……选对方式很重要
- 限流策略：别把人家服务器打爆了
- 重试机制：网络不稳定是常态
- 超时设置：别让程序死等
3.2 实战：带重试的API拉取
import requests
import time
from functools import wraps
def retry(max_retries=3, delay=1):
"""重试装饰器"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for i in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
print(f"第{i+1}次尝试失败: {e}")
if i < max_retries - 1:
time.sleep(delay * (i + 1)) # 指数退避
raise Exception("所有重试都失败了")
return wrapper
return decorator
@retry(max_retries=3, delay=2)
def fetch_device_config(device_id):
"""拉取设备配置"""
url = f"https://api.example.com/v1/devices/{device_id}/config"
headers = {"Authorization": "Bearer your_token_here"}
response = requests.get(url, timeout=5)
response.raise_for_status() # 非200状态码会抛异常
return response.json()
# 使用
try:
config = fetch_device_config("EDGE-001")
print(f"获取到配置: {config}")
except Exception as e:
print(f"配置拉取失败: {e}")我的习惯：API拉取的数据一定要做本地缓存。万一网络断了，设备还能用缓存数据继续工作。这叫「离线优先」设计。
四、数据格式解析：JSON/CSV/XML
数据拿到手了，但格式五花八门。JSON、CSV、XML是最常见的三种。说白了，就是要把字符串变成程序能处理的结构化数据。
4.1 三种格式对比
| 格式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| JSON | 轻量、易读、支持嵌套 | 无注释、不支持循环引用 | API数据、配置文件 |
| CSV | 简洁、表格化、Excel友好 | 不支持嵌套、类型信息丢失 | 传感器数据、日志导出 |
| XML | 结构化强、支持命名空间 | 冗长、解析慢 | 工业协议、遗留系统 |
4.2 实战：统一解析器
我做过一个项目，设备上报的数据格式不统一，有的用JSON，有的用CSV，还有老设备用XML。后来我写了个统一解析器，根据文件后缀自动选择解析方式。
import json
import csv
import xml.etree.ElementTree as ET
from io import StringIO
def parse_data(data, format_type):
"""统一数据解析入口"""
if format_type == 'json':
return json.loads(data)
elif format_type == 'csv':
reader = csv.DictReader(StringIO(data))
return [row for row in reader]
elif format_type == 'xml':
root = ET.fromstring(data)
result = {}
for child in root:
result[child.tag] = child.text
return result
else:
raise ValueError(f"不支持的格式: {format_type}")
# 示例
json_data = '{"temperature": 25.3, "humidity": 60.1}'
csv_data = "time,temperature,humidity\n2024-01-15,25.3,60.1\n2024-01-16,26.0,59.5"
xml_data = '<data><temperature>25.3</temperature><humidity>60.1</humidity></data>'
print("JSON解析:", parse_data(json_data, 'json'))
print("CSV解析:", parse_data(csv_data, 'csv'))
print("XML解析:", parse_data(xml_data, 'xml'))避坑指南：解析外部数据时一定要做校验。我曾经遇到过CSV文件里混入了特殊字符，导致整个解析器崩溃。后来我养成了习惯——解析前先做数据清洗，比如去掉不可见字符、检查字段数量是否一致。
五、本章知识体系
说了这么多，咱们用一张图来总结一下数据采集的核心逻辑。
这张图其实就说明白了：数据从三个源头来，经过采集层统一接入，再到解析层转换成标准格式，最后交给预处理模块。每一步都有坑，但每一步也都有套路。
好了，这一章的内容就到这儿。数据采集是基础中的基础，但也是最容易出问题的地方。下一章咱们聊聊数据清洗——你会发现，采集回来的数据，十有八九都是「脏」的。