企业官网建设流程全解析

万亿级日志数据的采集、清洗与存储：从Filebeat到ClickHouse的全链路实践

一、每秒百万行日志涌入，旧架构在第三个小时就崩溃了

当业务日志量从每天的 500GB 增长到 10TB 时，基于 Elasticsearch + Logstash + Kibana（ELK）的日志平台开始出现"数据断层"——日志采集延迟从 5 秒飙升到 30 分钟，ES 节点频繁 OOM，Kibana 打开一个 Dashboard 需要 2 分钟。

分析 ELK 架构的瓶颈后发现：Logstash 的 JVM 内存模型在大量并发写入时存在 GC 停顿、ES 的倒排索引在每个字段都建立索引导致写入放大 5-10 倍、Kibana 的聚合查询在 TB 级别数据上近乎不可用。每个组件都不是"不好"，只是它们面对的不是千万级而是万亿级数据。

新的架构选型是 Filebeat + Kafka + ClickHouse + Grafana——用 ClickHouse 的列式存储和向量化执行引擎替代 ES 的倒排索引，用 Filebeat 的轻量级采集替代 Logstash 的重量级处理，用 Kafka 做流量削峰。本文将详细记录这三步改造的完整过程。

二、全链路数据流架构设计

flowchart LR subgraph Production["业务服务器集群"] S1[App Server 1<br/>Filebeat] S2[App Server 2<br/>Filebeat] S3[App Server N<br/>Filebeat] end S1 --> K1 S2 --> K2 S3 --> K3 subgraph Kafka["Kafka 集群 (缓冲层)"] K1[Broker 1<br/>Partition 0-2] K2[Broker 2<br/>Partition 3-5] K3[Broker 3<br/>Partition 6-8] end K1 --> C1 K2 --> C2 K3 --> C3 subgraph Consumer["消费与写入"] C1[ClickHouse Writer 1] C2[ClickHouse Writer 2] C3[ClickHouse Writer 3] end C1 --> CH1 C2 --> CH2 C3 --> CH3 subgraph ClickHouse["ClickHouse 集群"] CH1[Node 1] CH2[Node 2] CH3[Node 3] end CH1 --> G CH2 --> G CH3 --> G G[Grafana Dashboard]

关键设计决策：

1. Filebeat 替代 Logstash：Filebeat 用 Go 编写，内存占用仅 30MB，而 Logstash 的 JVM 占用超过 1GB。在 500 台服务器上部署，这就是 500GB vs 15GB 的内存差异。

2. Kafka 作为缓冲层：当 ClickHouse 写入遇到背压（如正在做 Part Merge）时，Kafka 可以缓存数小时的数据。没有这个缓冲层，Filebeat 会因为下游不可用而丢弃数据。

3. ClickHouse 替代 ES：用 MergeTree 引擎 + 分区 + 排序键替代倒排索引，存储空间减少 80%，查询性能提升 5-10 倍。

三、全链路代码实现

3.1 Filebeat 配置

# filebeat.yml filebeat.inputs: - type: log enabled: true paths: - /var/log/app/*.log - /var/log/nginx/access.log # 多行日志合并 (Java 异常堆栈) multiline.pattern: '^\d{4}-\d{2}-\d{2}' multiline.negate: true multiline.match: after # 字段预处理 fields: service: "order-service" environment: "production" datacenter: "shanghai" fields_under_root: false # 解析 JSON 格式日志 json.keys_under_root: true json.add_error_key: true json.overwrite_keys: true # 输出到 Kafka output.kafka: hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"] topic: "app-logs" partition.round_robin: reachable_only: true required_acks: 1 compression: lz4 max_message_bytes: 1048576 # 日志记录与监控 logging.level: info logging.to_files: true logging.files: path: /var/log/filebeat name: filebeat.log keepfiles: 7

3.2 Kafka → ClickHouse 消费写入

#!/usr/bin/env python3 """Kafka 消费者：消费日志并批量写入 ClickHouse""" from kafka import KafkaConsumer from clickhouse_driver import Client import json import time from typing import List, Dict from datetime import datetime import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class LogWriter: """日志写入 ClickHouse 的消费者""" BATCH_SIZE = 100000 # 每批最多 10 万条 FLUSH_INTERVAL = 5 # 每 5 秒刷一次 MAX_RETRIES = 3 def __init__(self, kafka_brokers: List[str], ch_host: str): self.consumer = KafkaConsumer( 'app-logs', bootstrap_servers=kafka_brokers, group_id='ch-log-writer', auto_offset_reset='latest', enable_auto_commit=False, max_poll_records=50000, fetch_max_bytes=50 * 1024 * 1024, value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) self.ch = Client( host=ch_host, port=9000, database='logs', settings={ 'max_partitions_per_insert_block': 100, 'async_insert': 1, 'wait_for_async_insert': 0 } ) self.buffer = [] self.last_flush = time.time() def run(self): """主消费循环""" try: for msg in self.consumer: self.buffer.append(msg.value) # 达到批量大小或时间间隔时写入 if (len(self.buffer) >= self.BATCH_SIZE or time.time() - self.last_flush >= self.FLUSH_INTERVAL): self._flush() except Exception as e: logger.error(f"消费异常: {e}") self._flush() raise def _flush(self): """批量写入 ClickHouse""" if not self.buffer: return rows = [] for record in self.buffer: row = self._transform(record) if row: rows.append(row) if not rows: self.buffer = [] return # 带重试的写入 for attempt in range(self.MAX_RETRIES): try: self.ch.execute( "INSERT INTO app_logs VALUES", rows ) logger.info(f"成功写入 {len(rows)} 条日志") break except Exception as e: logger.error(f"写入失败 (尝试 {attempt + 1}): {e}") if attempt < self.MAX_RETRIES - 1: time.sleep(2 ** attempt) self.buffer = [] self.last_flush = time.time() # 手动提交 offset self.consumer.commit() def _transform(self, record: Dict) -> tuple: """日志格式转换与清洗""" try: ts = record.get('@timestamp', datetime.now().isoformat()) timestamp = datetime.fromisoformat( ts.replace('Z', '+00:00') ) return ( timestamp, record.get('service', 'unknown'), record.get('level', 'INFO'), record.get('logger', ''), record.get('message', ''), record.get('fields', {}).get('trace_id', ''), record.get('fields', {}).get('user_id', 0), record.get('fields', {}).get('duration_ms', 0), json.dumps(record.get('fields', {}), ensure_ascii=False) ) except Exception as e: logger.warning(f"日志转换失败: {e}") return None

3.3 ClickHouse 表结构

-- 1. 本地表（每个节点独立） CREATE TABLE app_logs_local ON CLUSTER 'log_cluster' ( timestamp DateTime, service LowCardinality(String), level LowCardinality(String), logger String, message String, trace_id String, user_id UInt64, duration_ms UInt32, extra String ) ENGINE = MergeTree() PARTITION BY toYYYYMMDD(timestamp) ORDER BY (service, level, timestamp) TTL timestamp + INTERVAL 30 DAY -- 30天自动过期 SETTINGS index_granularity = 8192; -- 2. 分布式表（查询入口） CREATE TABLE app_logs ON CLUSTER 'log_cluster' AS app_logs_local ENGINE = Distributed('log_cluster', 'logs', 'app_logs_local', rand());

四、三条架构边界的警醒

边界一：Kafka 的"无限缓冲"是幻觉

Kafka 的 Topic 大小由磁盘容量决定。当消费者故障超过 Kafka 的保留时间（默认 7 天），数据将永久丢失。因此 Kafka 不是"无限缓冲"，而是"有限缓冲"。

边界二：ClickHouse 的写入峰值

ClickHouse 的最大写入吞吐受 Part Merge 速度限制。当 Merge 速度跟不上 Insert 速度时，Parts 数量爆炸（>10000），查询性能急剧下降。解决方案是控制写入批次大小和合理分区。

边界三：日志清洗在"哪里做"

在 Filebeat 端做清洗（Processor）可以减少无效数据进入 Kafka，但会消耗业务服务器资源。在消费端做清洗可以利用消费者组的并行能力。实践中在 Filebeat 端做轻量过滤，在消费者端做复杂清洗是最佳平衡。

五、总结

万亿级日志系统的改造，表面上是工具替换（ELK → Kafka + ClickHouse），实质上是架构思想的转变：

1. 解耦采集与存储：Filebeat 负责采集，Kafka 负责缓冲，ClickHouse 负责存储——每层职责单一
2. 列式存储是日志的天配：日志查询多为"按服务、时间、级别聚合"，ClickHouse 的列存天然适合这类 OLAP 查询
3. TTL 自动过期是运维的福报：不需要手动清理过期日志，MergeTree Engine 的 TTL 机制自动完成

改造后，日志平台的 TCO（总拥有成本）降低了 67%，主要是存储成本（ES 的倒排索引开销极大）和服务器成本（Logstash 的 JVM 需求）。查询响应时间从分钟级降到秒级，运维团队终于可以在故障时实时查看日志而非等待 Kibana 加载。