万亿级日志数据的采集、清洗与存储从Filebeat到ClickHouse的全链路实践一、每秒百万行日志涌入旧架构在第三个小时就崩溃了当业务日志量从每天的 500GB 增长到 10TB 时基于 Elasticsearch Logstash KibanaELK的日志平台开始出现数据断层——日志采集延迟从 5 秒飙升到 30 分钟ES 节点频繁 OOMKibana 打开一个 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 1br/Filebeat] S2[App Server 2br/Filebeat] S3[App Server Nbr/Filebeat] end S1 -- K1 S2 -- K2 S3 -- K3 subgraph Kafka[Kafka 集群 (缓冲层)] K1[Broker 1br/Partition 0-2] K2[Broker 2br/Partition 3-5] K3[Broker 3br/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]关键设计决策Filebeat 替代 LogstashFilebeat 用 Go 编写内存占用仅 30MB而 Logstash 的 JVM 占用超过 1GB。在 500 台服务器上部署这就是 500GB vs 15GB 的内存差异。Kafka 作为缓冲层当 ClickHouse 写入遇到背压如正在做 Part Merge时Kafka 可以缓存数小时的数据。没有这个缓冲层Filebeat 会因为下游不可用而丢弃数据。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: 73.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(levellogging.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_serverskafka_brokers, group_idch-log-writer, auto_offset_resetlatest, enable_auto_commitFalse, max_poll_records50000, fetch_max_bytes50 * 1024 * 1024, value_deserializerlambda m: json.loads(m.decode(utf-8)) ) self.ch Client( hostch_host, port9000, databaselogs, 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_asciiFalse) ) except Exception as e: logger.warning(f日志转换失败: {e}) return None3.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实质上是架构思想的转变解耦采集与存储Filebeat 负责采集Kafka 负责缓冲ClickHouse 负责存储——每层职责单一列式存储是日志的天配日志查询多为按服务、时间、级别聚合ClickHouse 的列存天然适合这类 OLAP 查询TTL 自动过期是运维的福报不需要手动清理过期日志MergeTree Engine 的 TTL 机制自动完成改造后日志平台的 TCO总拥有成本降低了 67%主要是存储成本ES 的倒排索引开销极大和服务器成本Logstash 的 JVM 需求。查询响应时间从分钟级降到秒级运维团队终于可以在故障时实时查看日志而非等待 Kibana 加载。