4. 原始数据收集：传感器数据、日志数据、API 响应、文本数据的收集策略

做特征学习，第一步不是调模型，而是把原始数据老老实实收上来。我见过太多团队在数据收集阶段就埋了雷——漏字段、时间错位、重复行，后期怎么洗都洗不干净。说白了，收集阶段的策略决定了整个管线的上限。

这章我按四种常见数据类型，把实际生产中常用的收集策略拆开讲。每种我都踩过坑，你想想看，有些坑其实花十分钟就能避开。

4.1 传感器数据收集

传感器数据的特点是高频、连续、异构。我在项目里处理过一个工业 IoT 场景，温度、振动、电流三个传感器，采样率从 10Hz 到 1000Hz 不等。最头疼的是时间戳对齐——不同传感器用不同的本地时钟，差了毫秒级都会导致特征计算出错。

收集策略的核心就三点：

• 协议适配：低频设备用 MQTT，高频设备用 gRPC 流。MQTT 的 QoS 级别一定要设对，QoS 0 丢数据，QoS 2 太慢，我一般设 QoS 1。
• 时间戳归一：所有数据到达后立即打上接收端时间戳，保留原始设备时间戳作为辅助。
• 去重与排序：网络重传会导致乱序和重复，用 Redis 的 sorted set 做窗口排序。

一个简单的 MQTT 消费者代码片（Python 环境）：

import paho.mqtt.client as mqtt
import time

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    # 立即打上接收时间戳
    record = {"ts_recv": time.time_ns(), "topic": msg.topic, "data": payload}
    # 写入消息队列（Kafka / Redis）
    queue.put(record)

client = mqtt.Client()
client.on_message = on_message
client.connect("broker.local", 1883, 60)
client.subscribe("sensors/#", qos=1)
client.loop_forever()

4.2 日志数据收集

日志是最容易「忽略规模」的数据类型。以前我把 Nginx 日志直接写文件，一个月后磁盘爆了——单台机器一天就能产出几个 GB。收集日志时，轮转、压缩、异步采集是标配。

我用 Filebeat → Kafka → Logstash 这套组合最多。Filebeat 负责监听文件变化，不会丢行；Kafka 做缓冲，扛流量冲击；Logstash 做解析和过滤。嗯，这里要注意：日志格式一定要统一，否则 Logstash 的 grok 表达式会写哭你。

Filebeat 配置示例（保持原样）：

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/app/*.log
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true
  multiline.match: after

output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092"]
  topic: "app-logs"
  compression: gzip
  max_message_bytes: 1048576

4.3 API 响应数据收集

API 数据收集最怕两件事：被限流，和收不全。我在做电商用户浏览行为采集时，第三方推荐 API 限制 QPS 为 10，但我们需要每秒打几千次。怎么办？加本地队列+令牌桶。

策略要点：

• 重试与退避：指数退避 + jitter，避免惊群效应。
• 增量拉取：使用时间戳或偏移量字段，每次只取增量，不重复。
• 数据校验：API 返回的 JSON 可能漏字段，一定要做 schema 校验。

带重试的 Python 收集示例：

import requests
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=2, max=30))
def fetch_page(url, params):
    resp = requests.get(url, params=params, timeout=10)
    resp.raise_for_status()
    return resp.json()

# 使用游标分页
cursor = None
while True:
    params = {"limit": 1000}
    if cursor:
        params["cursor"] = cursor
    data = fetch_page("https://api.example.com/events", params)
    # 校验 schema
    for item in data.get("items", []):
        assert "id" in item and "timestamp" in item
    # 批量写入存储
    write_to_storage(data["items"])
    cursor = data.get("next_cursor")
    if not cursor:
        break

4.4 文本数据收集

文本数据源五花八门：网页、PDF、Markdown 文档、用户评论。我做过一个舆情分析项目，每天爬 10 万个微信公众号文章，最坑的是反爬策略和编码问题。

收集策略很简单但容易漏：

• 明确边界：正文范围、发布时间、作者等字段要精确定位。不要傻抓整个 DOM。
• 编码统一：全部转成 UTF-8 后再存储。GBK、Big5 混在一起会出鬼。
• 去重：文本相似去重比 URL 去重重要。很多文章换 URL 但内容一样。

简单的正文提取逻辑示意：

from bs4 import BeautifulSoup
import requests

url = "https://example.com/article"
html = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}).text
soup = BeautifulSoup(html, "html.parser")

# 移除脚本和样式
for tag in soup(["script", "style"]):
    tag.decompose()

text = soup.get_text(separator="\n")
# 进一步清理空行和短行
lines = [line.strip() for line in text.splitlines() if len(line.strip()) > 10]
clean_text = "\n".join(lines)
encoding = soup.original_encoding
# 统一为 UTF-8
clean_text = clean_text.encode("utf-8", errors="ignore").decode("utf-8")

4.5 统一收集架构与对比

别看数据源五花八门，最终的收集管道其实是相似的。我个人习惯先画一张图，把数据流理清楚：

最后用一个表把这四种数据的核心差异总结一下，方便你快速对照选型：