第21课：日志分析管道

📋 日志分析管道

构建完整的日志收集→解析→存储→分析管道，从访问日志中提取业务洞察。

🐍 Python实战：核心实战

# 日志分析管道
import re, time, random, json
from datetime import datetime, timedelta
from collections import defaultdict

class LogGenerator:
    def __init__(self):
        self.ips = [f"192.168.{random.randint(1,255)}.{random.randint(1,255)}" for _ in range(50)]
        self.paths = ['/','/api/users','/api/orders','/products','/search','/login','/checkout']
        self.codes = [200,200,200,200,301,404,500]
    def generate(self, count=1000):
        logs = []
        for _ in range(count):
            ts = datetime.now() - timedelta(seconds=random.randint(0,3600))
            log = f'{self.ips[random.randint(0,len(self.ips)-1)]} - - [{ts.strftime("%d/%b/%Y:%H:%M:%S")}] "GET {random.choice(self.paths)} HTTP/1.1" {random.choice(self.codes)} {random.randint(100,50000)}'
            logs.append(log)
        return logs

class LogParser:
    PATTERN = re.compile(r'(\S+) - - \[([^\]]+)\] "(\w+) (\S+) \S+" (\d+) (\d+)')
    def parse(self, log_line):
        m = self.PATTERN.match(log_line)
        if m:
            return {'ip':m.group(1),'timestamp':m.group(2),'method':m.group(3),'path':m.group(4),'status':int(m.group(5)),'size':int(m.group(6))}
        return None

class LogAnalyzer:
    def __init__(self): self.stats = defaultdict(lambda: defaultdict(int))
    def analyze(self, parsed_logs):
        for log in parsed_logs:
            self.stats['status'][log['status']] += 1
            self.stats['path'][log['path']] += 1
            self.stats['ip'][log['ip']] += 1
    def top_n(self, category, n=5):
        return sorted(self.stats[category].items(), key=lambda x: x[1], reverse=True)[:n]
    def error_rate(self):
        total = sum(self.stats['status'].values())
        errors = sum(v for k,v in self.stats['status'].items() if k >= 400)
        return round(errors/total*100,2) if total else 0
    def qps(self, duration_sec=3600):
        total = sum(self.stats['status'].values())
        return round(total/duration_sec, 2)

# 运行
gen = LogGenerator()
parser = LogParser()
analyzer = LogAnalyzer()

logs = gen.generate(5000)
parsed = [p for log in logs if (p := parser.parse(log))]
analyzer.analyze(parsed)

print("日志分析管道:")
print(f"  总请求: {len(parsed)}")
print(f"  错误率: {analyzer.error_rate()}%")
print(f"  QPS: {analyzer.qps()}")
print(f"  Top路径: {analyzer.top_n('path',3)}")
print(f"  Top IP: {analyzer.top_n('ip',3)}")
print(f"  状态码: {dict(analyzer.stats['status'])}")
print("✅ 验证通过 - 日志分析管道运行正常")

📖 原理深入：企业数据平台架构

🏆 最佳实践：数据平台建设检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答

🔑 关键技术选型决策

数据量级？
├── <1GB/天 → SQLite/PostgreSQL + Python脚本
├── 1-100GB/天 → 数据仓库(Snowflake/BQ) + dbt
├── 100GB-1TB/天 → Spark + 数据湖
└── >1TB/天 → Spark/Flink + 湖仓一体

延迟要求？
├── T+1 → 批处理(Spark) + Airflow
├── 小时级 → 微批(Spark Streaming) + Airflow
├── 分钟级 → 流处理(Flink) + Kafka
└── 秒级 → 纯流(Flink) + Kafka + 缓存

数据类型？
├── 纯结构化 → 数据仓库
├── 结构化+半结构化 → 湖仓一体
└── 全类型 → 数据湖 + 元数据管理