数据采集基础：数据源接入与预处理
数据采集是整个新鲜度监控的根基。我见过太多团队在阈值调优上花了大把精力，最后发现是底层数据质量有问题——白忙一场。这一章，咱们就把数据采集这块地基打牢。
数据源接入：三种主流方式
实际生产环境中，数据源无非三种：API、数据库、日志文件。每种方式都有各自的坑，我一个个说。
1. API 接入
API 是最灵活的方式，但也是最容易出幺蛾子的。我记得有一次对接某个第三方监控平台，对方 API 返回的数据格式说变就变，连个通知都没有。
核心要点：API 接入必须做好三件事——重试机制、限流保护、格式校验。
# 一个带重试和限流的 API 采集器示例
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
import time
class APICollector:
def __init__(self, endpoint, , rate_limit=10):
self.endpoint = endpoint
self.headers = {'Authorization': f'Bearer {}'}
self.rate_limit = rate_limit # 每秒请求数
self.last_request = 0
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_metrics(self):
# 限流控制
elapsed = time.time() - self.last_request
if elapsed < 1.0 / self.rate_limit:
time.sleep(1.0 / self.rate_limit - elapsed)
resp = requests.get(self.endpoint, headers=self.headers, timeout=5)
resp.raise_for_status()
self.last_request = time.time()
return resp.json()我曾经踩过的坑：API 返回的 JSON 字段名可能带大小写混用、下划线、驼峰混着来。接入时一定要做字段名规范化，否则后面清洗阶段会疯掉。
2. 数据库接入
数据库接入看起来简单，但性能问题往往藏得很深。你想想看，如果直接从业务库拉全量数据，那 DBA 不找你喝茶才怪。
| 接入方式 | 适用场景 | 注意事项 |
|---|---|---|
| 全量拉取 | 小表（<10万行） | 避免高峰期执行 |
| 增量拉取 | 大表、实时性要求高 | 需要时间戳或自增ID |
| CDC（变更数据捕获） | 实时流式接入 | 需要数据库开启 binlog |
我个人习惯用增量拉取配合 CDC。增量拉取简单可靠，CDC 适合对实时性要求高的场景。但要注意，CDC 会额外消耗数据库 IO，别在生产库上直接开。
3. 日志文件接入
日志接入，说白了就是 tail -f 的升级版。但生产环境里，日志轮转、文件锁、编码问题，哪个都能让你头疼半天。
我的经验：日志采集最好用 Filebeat 或 Fluentd 这类成熟工具，别自己造轮子。它们处理文件轮转、断点续传、背压控制都很成熟。
# 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"
partition.round_robin:
reachable_only: true数据清洗与预处理
数据进了采集层，不等于就能直接用。脏数据、异常值、格式不一致，这些都得在预处理阶段干掉。
常见脏数据类型
- 缺失值：字段为空或 null，需要填充或丢弃
- 异常值：超出合理范围的数据点，比如 CPU 使用率 150%
- 重复数据：同一时间点的重复采集，需要去重
- 格式错误：时间戳格式不统一、数值带单位等
核心原则：清洗规则宁可严格，不要宽松。放过一条脏数据，后面整个阈值模型都可能被带偏。
# 数据清洗示例
import pandas as pd
import numpy as np
def clean_metrics(df):
# 1. 去除重复数据
df = df.drop_duplicates(subset=['metric_name', 'timestamp'])
# 2. 处理缺失值
df = df.fillna(method='ffill') # 前向填充
# 3. 过滤异常值（以 CPU 使用率为例）
df = df[(df['value'] >= 0) & (df['value'] <= 100)]
# 4. 统一时间戳格式
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='s')
return df时间窗口对齐
这是数据采集里最容易出问题的一环。不同数据源的时间粒度不一样——API 可能 10 秒一次，数据库 1 分钟一次，日志又是毫秒级的。怎么对齐？
对齐策略
| 策略 | 做法 | 适用场景 |
|---|---|---|
| 降采样 | 将细粒度数据聚合到粗粒度 | 多数场景，推荐使用 |
| 升采样 | 将粗粒度数据插值到细粒度 | 需要高精度时，慎用 |
| 滑动窗口 | 固定时间窗口内聚合 | 实时监控场景 |
我一般用降采样配合滑动窗口。举个例子，日志是毫秒级的，API 是 10 秒级的，那就统一对齐到 1 分钟窗口，取平均值或百分位数。
注意：时间窗口对齐时，一定要处理好边界情况。比如窗口边界的数据点归属问题，我见过有人把 10:00:00 的数据算到上一个窗口，结果整个趋势图都偏移了。
# 时间窗口对齐示例
def align_time_windows(df, window_size='1min'):
# 设置时间戳为索引
df = df.set_index('timestamp')
# 按窗口聚合，取平均值
aligned = df.resample(window_size).mean()
# 处理空窗口（前向填充）
aligned = aligned.fillna(method='ffill', limit=3)
return aligned.reset_index()整体数据采集流程
下面这张图，把整个数据采集的脉络串起来了。从数据源到清洗对齐，每一步都有讲究。
嗯，数据采集这块就这些。记住一个原则：宁可采集多，不要采集少；宁可清洗严，不要清洗松。后面阈值调优的效果，很大程度上取决于这一步做得怎么样。