2. 数据采集与清洗：数据源接入与异常处理
数据采集与清洗，说白了就是给成本分析系统「喂饭」。饭喂得不好，后面所有可视化图表都是垃圾。我在好几个项目里都见过这种情况——数据源没接对，清洗规则没定好，结果预警系统天天误报，最后没人信了。
这一章，我们就聊聊怎么把数据从各个地方捞进来，怎么洗干净，怎么处理那些烦人的异常值。
2.1 数据源接入：API、数据库、日志
成本数据通常来自三个地方：API接口、数据库、日志文件。每种方式都有它的脾气，我们得顺着来。
2.1.1 API 接入
云厂商的账单接口，基本都是 RESTful API。比如 AWS 的 Cost Explorer、阿里云的 Billing API。我个人习惯先写一个封装层，把认证、分页、重试逻辑统一处理掉。
import requests
import time
class CostAPIClient:
def __init__(self, , base_url):
self. =
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {self.}',
'Content-Type': 'application/json'
})
def fetch_cost_data(self, start_date, end_date, max_retries=3):
url = f'{self.base_url}/v1/cost/usage'
params = {
'start': start_date,
'end': end_date,
'granularity': 'DAILY'
}
for attempt in range(max_retries):
try:
resp = self.session.get(url, params=params, timeout=30)
resp.raise_for_status()
return resp.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt) # 指数退避
2.1.2 数据库接入
很多公司会把成本数据先存到自己的数据库里，比如 MySQL、PostgreSQL。这时候我们需要做增量同步，而不是每次都全量拉取。
| 同步方式 | 适用场景 | 注意事项 |
|---|---|---|
| 全量同步 | 数据量小（<10万行） | 每次覆盖，简单但慢 |
| 增量同步 | 数据量大，持续增长 | 需要时间戳或自增ID |
| CDC（变更数据捕获） | 实时性要求高 | 需要数据库开启binlog |
我建议优先用增量同步。你想想看，如果每天几百万条账单记录都全量拉一遍，数据库扛不住，网络带宽也受不了。
-- 增量同步示例：只拉取最近7天的数据
SELECT
resource_id,
cost_amount,
currency,
usage_date,
region
FROM cost_records
WHERE usage_date >= DATE_SUB(CURDATE(), INTERVAL 7 DAY)
AND sync_status = 0; -- 未同步标记
2.1.3 日志接入
日志文件是另一个重要来源。比如 Nginx 访问日志、应用日志里，可能埋了资源使用量的数据。我曾经接手过一个项目，成本数据全在几百台服务器的日志文件里，每天几十GB。
处理日志，我推荐用 Filebeat + Logstash 的组合。Filebeat 负责采集，Logstash 负责解析和清洗。
# Logstash 配置示例
input {
beats {
port => 5044
}
}
filter {
grok {
match => {
"message" => "%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{DATA:service} cost=%{NUMBER:cost:float}"
}
}
date {
match => ["timestamp", "ISO8601"]
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "cost-logs-%{+YYYY.MM.dd}"
}
}
2.2 数据清洗规则
数据接进来了，但往往不干净。常见的脏数据有：字段缺失、格式不统一、重复记录、单位不一致。我们需要制定清洗规则。
2.2.1 字段标准化
同一个意思，不同系统可能用不同字段名。比如「费用」在A系统叫 cost，在B系统叫 amount，在C系统叫 charge。我们需要统一映射。
# 字段映射规则
FIELD_MAPPING = {
'cost': 'cost_amount',
'amount': 'cost_amount',
'charge': 'cost_amount',
'resource_id': 'resource_id',
'res_id': 'resource_id',
'rid': 'resource_id'
}
def normalize_fields(record):
normalized = {}
for old_key, value in record.items():
new_key = FIELD_MAPPING.get(old_key, old_key)
normalized[new_key] = value
return normalized
2.2.2 单位统一
成本数据里，单位不一致是个大坑。有的用元，有的用分，有的用美元。我建议统一转换成「元（人民币）」作为标准单位。
| 原始单位 | 转换规则 | 示例 |
|---|---|---|
| 分（人民币） | 除以100 | 100分 → 1元 |
| 美元 | 乘以汇率 | 10美元 → 约72元 |
| 厘 | 除以1000 | 1000厘 → 1元 |
2.3 异常值处理
异常值处理，是数据清洗里最考验经验的部分。处理不好，预警系统就会乱叫。
2.3.1 常见的异常类型
- 负值异常：成本出现负数，可能是退款、折扣、或者数据错误
- 突增突降：某天成本突然暴涨10倍，可能是计费周期调整
- 空值/零值：某些资源没有成本数据，可能是未使用或采集遗漏
- 重复记录：同一条账单被采集了多次
2.3.2 异常检测方法
我个人习惯用「3σ原则」来检测数值异常。简单说，就是计算历史数据的均值和标准差，超出均值±3倍标准差的值，标记为异常。
import numpy as np
def detect_anomalies(cost_series, threshold=3):
"""
使用3σ原则检测异常值
"""
mean = np.mean(cost_series)
std = np.std(cost_series)
anomalies = []
for idx, value in enumerate(cost_series):
z_score = (value - mean) / std
if abs(z_score) > threshold:
anomalies.append({
'index': idx,
'value': value,
'z_score': z_score,
'reason': f'超出{threshold}σ范围'
})
return anomalies
2.3.3 异常处理策略
检测到异常后，怎么处理？我总结了四种策略：
- 自动修正：对于明显的数据错误（如负值），用历史均值替换
- 标记待审：对于不确定的异常，打上标签，人工审核
- 忽略通过：对于已知的合理波动（如月初账单），直接放行
- 阻断告警：对于严重异常，直接触发告警，停止后续流程
def handle_anomaly(record, anomaly_type):
if anomaly_type == 'negative_cost':
# 自动修正：用最近7天均值替换
avg_cost = get_recent_avg_cost(record['resource_id'], days=7)
record['cost_amount'] = avg_cost
record['cleaning_flag'] = 'auto_fixed'
record['original_value'] = record['cost_amount']
elif anomaly_type == 'spike':
# 标记待审
record['cleaning_flag'] = 'pending_review'
send_review_notification(record)
elif anomaly_type == 'known_pattern':
# 忽略通过
record['cleaning_flag'] = 'ignored'
return record
2.4 数据采集与清洗的整体流程
说了这么多，我们来看一张整体流程图。它把数据从源头到清洗完成的全过程串起来了。
这张图把整个流程分成了五层：数据源、采集层、清洗层、异常处理层、输出层。每一层都有明确的职责，层与层之间通过标准接口传递数据。这样做的好处是——哪一层出了问题，可以快速定位修复，不影响其他层。