4. 日志采集与结构化处理实战
数据平台通常由多种组件构成：EMR 上的 Spark 作业、Glue ETL 任务、RDS 数据库、Lambda 函数、API Gateway 访问日志、容器化服务等。这些组件各自把日志写到 CloudWatch Logs 的不同日志组中，格式不统一、字段不完整、保留周期短，排障时需要在多个控制台之间来回切换。本节的目标是把这些分散日志集中采集、解析、富化并落到可长期查询的存储中，形成一条可复用的日志管道。
一、整体管道架构与选型
一条典型的集中式日志管道由四段组成：采集端（CloudWatch Logs 订阅过滤）、传输与缓冲（Kinesis Data Streams 或直接 Firehose）、转换（Lambda 解析与富化）、投递（OpenSearch 或 S3）。选择哪条路径取决于查询模式：需要近实时全文检索与仪表盘时走 OpenSearch；需要低成本长期留存、按需用 Athena 做 SQL 分析时走 S3。
CloudWatch Logs 的订阅过滤器（Subscription Filter）是采集入口。它把匹配过滤模式的日志事件以 gzip 压缩、Base64 编码的 JSON 记录批量推送到目标。目标可以是 Kinesis Data Streams、Kinesis Data Firehose 或 Lambda。生产上更推荐先进入 Kinesis Data Streams，再由 Firehose 消费，这样可以在流上做多消费者、重放和背压控制；如果只是单一投递路径，也可以让订阅过滤器直接指向 Firehose。
二、采集端：订阅过滤器与过滤模式
订阅过滤器通过 Filter Pattern 决定哪些日志事件被推送。Filter Pattern 支持空格分隔的多个词（AND 关系）、问号表示可选、双引号表示精确短语、? 与 - 排除等语法。例如只采集 ERROR 与 WARN 级别日志，可用模式 ?ERROR ?WARN；若要排除健康检查噪声，可用 -healthcheck。过滤模式写得越精确，下游 Firehose 与 Lambda 的成本越低。
创建订阅过滤器时，需要给 CloudWatch Logs 一个可代入的 IAM 角色，允许其调用 firehose:PutRecordBatch 或 kinesis:PutRecords。同时注意每个日志组默认最多 2 个订阅过滤器，且同一日志组不能同时向同一目标重复推送。若日志量很大，建议在日志组层面先做采样或分级，避免把 DEBUG 日志全部灌入管道。
三、传输与缓冲：Kinesis Data Streams 与 Firehose
Kinesis Data Streams 提供分片级的有序缓冲，适合需要多消费者（例如同时投递 OpenSearch 和 S3）或需要重放历史数据的场景。分片数量按写入吞吐估算：每个分片支持 1 MB/s 或 1000 条记录/s 的写入。假设某数据平台日志峰值约 3 MB/s，则至少需要 4 个分片留出余量。
Firehose 负责把流数据批量投递到目标。它的关键参数是缓冲大小与缓冲间隔：两者满足其一即触发投递。对 OpenSearch 目标，通常设置 Buffer Size 5 MB、Interval 60 s，以平衡延迟与请求数；对 S3 目标，可设置 128 MB、300 s，减少小文件。Firehose 还支持在投递前调用 Lambda 做同步转换，这就是下一节的转换环节。
四、转换：Lambda 解析与富化
Firehose 的 Lambda 转换会对每条记录调用一次函数，输入是 Base64 编码的记录，输出需返回 recordId、result（Ok/Dropped/ProcessingFailed）和 data。下面是一个把 CloudWatch Logs 原始事件解析为结构化 JSON 的示例：
import json, base64, gzip, re
def lambda_handler(event, context):
output = []
for record in event['records']:
raw = base64.b64decode(record['data'])
# CloudWatch Logs 订阅数据为 gzip 压缩
try:
payload = json.loads(gzip.decompress(raw))
except Exception:
payload = json.loads(raw)
for log_event in payload.get('logEvents', []):
msg = log_event['message']
# 示例：解析 Spark 日志中的级别、时间、类名
m = re.match(r'(\d{2}/\d{2}/\d{2} \d{2}:\d{2}:\d{2}) (\w+) ([\w\.]+): (.*)', msg)
if m:
ts, level, cls, text = m.groups()
enriched = {
'timestamp': ts,
'level': level,
'logger': cls,
'message': text,
'logGroup': payload.get('logGroup'),
'logStream': payload.get('logStream'),
'env': 'prod'
}
else:
enriched = {'level': 'UNKNOWN', 'message': msg,
'logGroup': payload.get('logGroup')}
output.append({
'recordId': record['recordId'],
'result': 'Ok',
'data': base64.b64encode(
(json.dumps(enriched) + '\n').encode()).decode()
})
return {'records': output}
转换中常见的富化操作包括：补全环境标签（env、region）、把日志组名映射为服务名、对手机号或身份证做脱敏、把时间戳统一为 ISO 8601。若某条记录解析失败，应返回 Dropped 而不是抛异常，否则整批记录会被重试并可能阻塞管道。建议同时配置 Firehose 的 S3 备份桶，把原始记录与转换后记录分别留存，便于回溯。
五、投递与查询：OpenSearch 与 S3+Athena
OpenSearch 适合近实时检索。Firehose 投递时会按索引名（如 logs-prod-2025.06）写入，索引名支持日期表达式。写入前需在 OpenSearch 中创建索引模板，显式定义 level、logger、timestamp 等字段类型，避免动态映射把时间戳识别为 text 导致无法按时间范围过滤。Kibana 中可基于 level:ERROR 与 logger 聚合快速定位异常来源。
S3+Athena 路径则把转换后的 JSON 或 Parquet 按 year=/month=/day=/hour= 分区写入。Athena 建表时使用分区投影或 MSCK REPAIR 加载分区，即可用标准 SQL 做跨服务日志关联分析。例如统计某小时内各服务的 ERROR 数量：
SELECT service, count(*) AS err_cnt
FROM logs_db.platform_logs
WHERE year='2025' AND month='06' AND day='15' AND hour='10'
AND level='ERROR'
GROUP BY service
ORDER BY err_cnt DESC;
两种存储并非互斥：常见做法是同时投递到 OpenSearch（近 7 天热数据）和 S3（长期冷数据），用同一份 Lambda 转换输出，保证字段一致。
六、验证与排障要点
管道上线后，先确认订阅过滤器已生效：在日志组中写入一条测试日志，观察 Kinesis 的 IncomingRecords 指标是否上升。若 Firehose 的 DeliveryToOpenSearch.Success 为 0，检查目标域权限与索引模板；若 Lambda 转换报错，查看 Firehose 的 ExecuteProcessing.Failure 指标与 Lambda 的 CloudWatch 日志。常见问题包括：Base64 解码后未解 gzip、返回结果缺少 recordId、单条记录超过 1 MB 限制、OpenSearch 字段类型冲突。建议为 Firehose 配置 CloudWatch 告警，监控 DeliveryToS3.DataFreshness 与 DeliveryToOpenSearch.DataFreshness，超过阈值即通知。