中间件--ClickHouse-12--案例-1-日志分析和监控

1、案例背景

一家互联网公司需要实时分析其服务器日志、应用日志和用户行为日志，以快速发现潜在问题并优化系统性能。

2、需求分析

目标：实时分析日志数据，快速发现问题并优化系统性能。
数据来源：
- 服务器日志：如 Nginx、Tomcat、Docker等日志。
- 应用日志：业务系统的运行日志。
- 用户行为日志：用户操作记录（如点击、浏览、下单等）。
输出：
- 错误率、请求延迟、用户行为路径等关键指标。
- 实时监控仪表盘。

3、解决思路

日志采集：使用工具（如Filebeat或Fluentd）将日志数据写入Kafka。
数据存储与分析：Kafka中的数据被导入到ClickHouse，利用其高效的压缩和查询性能进行日志分析。
可视化：通过 Grafana 或 Redash 构建仪表盘，展示关键指标（如错误率、请求延迟等）。

4、技术选型

日志采集：Filebeat 或 Fluentd。
消息队列：Kafka（用于缓冲和传输日志数据）。
存储与分析：ClickHouse（高性能 OLAP 数据库）。
可视化：Grafana 或 Redash。

5、ClickHouse的作用

高效存储：日志数据量通常非常庞大，ClickHouse的列式存储和高压缩比显著降低了存储成本。
实时分析：支持毫秒级响应的复杂查询，适合对海量日志进行实时分析。
灵活扩展：支持分布式部署，能够处理PB级别的日志数据。

6、基本实现步骤

（1）、日志采集

1. 安装 Filebeat：

bash示例：

sudo apt-get install filebeat

2. 配置 Filebeat：

编辑 filebeat.yml 文件，指定日志文件路径和 Kafka 输出：
yaml示例：

  filebeat.inputs:- type: logpaths:- /var/log/nginx/*.log- /var/log/application/*.logoutput.kafka:hosts: ["kafka-broker:9092"]topic: "logs"

解释：
Input为采集日志相关配置，如nginx的log日志文件，应用程序的log日志文件，output指定输出到kafka。

3. 启动 Filebeat：

bash示例：

sudo service filebeat start

（2）、消息队列（Kafka）

1. 安装 Kafka：

bash示例：

wget https://downloads.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
tar -xzf kafka_2.13-3.0.0.tgz
cd kafka_2.13-3.0.0

2. 启动 Kafka：

bash示例：

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties

（3）、数据消费与写入ClickHouse

1. 创建 ClickHouse 表：

sql示例：

   CREATE TABLE logs (timestamp DateTime,level String,message String,source String) ENGINE = MergeTree()ORDER BY (timestamp);

（4）、可视化

1. 安装 Grafana：

bash示例：

sudo apt-get install grafana
sudo service grafana-server start

2. 配置 ClickHouse 数据源：

在 Grafana 中添加 ClickHouse 数据源，配置连接信息。

3. 构建仪表盘：

创建图表展示日志的关键指标，如错误率、请求延迟等。

7、Spring Boot代码示例

使用Spring Boot消费Kafka数据并写入 ClickHouse。

（1）、添加依赖

<dependencies><!-- Spring Boot Starter --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><!-- Kafka --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><!-- ClickHouse JDBC --><dependency><groupId>com.clickhouse</groupId><artifactId>clickhouse-jdbc</artifactId><version>0.3.2</version></dependency><!-- Jackson for JSON Parsing --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId></dependency>
</dependencies>

（2）、配置 Kafka 和 ClickHouse

在 application.yml 中配置 Kafka 和 ClickHouse：

spring:kafka:bootstrap-servers: kafka-broker:9092consumer:group-id: clickhouse-groupauto-offset-reset: earliestdatasource:url: jdbc:clickhouse://clickhouse-server:8123/defaultdriver-class-name: com.clickhouse.jdbc.ClickHouseDriverusername: defaultpassword:

（3）、Kafka 消费者

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;@Service
public class LogConsumer {private final LogRepository logRepository;public LogConsumer(LogRepository logRepository) {this.logRepository = logRepository;}@KafkaListener(topics = "logs", groupId = "clickhouse-group")public void consume(String message) {// 解析日志消息Log log = parseLog(message);// 写入 ClickHouselogRepository.save(log);}private Log parseLog(String message) {// 假设日志是 JSON 格式ObjectMapper objectMapper = new ObjectMapper();try {return objectMapper.readValue(message, Log.class);} catch (Exception e) {throw new RuntimeException("Failed to parse log message", e);}}
}

（4）、ClickHouse 数据访问层

创建Repository类。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;@Repository
public class LogRepository {private final JdbcTemplate jdbcTemplate;@Autowiredpublic LogRepository(JdbcTemplate jdbcTemplate) {this.jdbcTemplate = jdbcTemplate;}public void save(Log log) {String sql = "INSERT INTO logs (timestamp, level, message, source) VALUES (?, ?, ?, ?)";jdbcTemplate.update(sql, log.getTimestamp(), log.getLevel(), log.getMessage(), log.getSource());}
}

（5）、日志实体类

import java.time.LocalDateTime;public class Log {private LocalDateTime timestamp;private String level;private String message;private String source;// Getters and Setters
}

（6）、 Service 层（LogService.java）

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.Map;@Service
public class LogService {@Autowiredprivate JdbcTemplate jdbcTemplate;// 查询最近5分钟的错误率public List<Map<String, Object>> getErrorRate() {String sql = "SELECT program, error_count * 100.0 / total_requests AS error_percent " +"FROM log_errors_mv " +"WHERE minute >= now() - interval 5 minute " +"GROUP BY program";return jdbcTemplate.queryForList(sql);}// 查询指定时间段的响应时间分布public List<Map<String, Object>> getResponseTimeStats(String startTime, String endTime) {String sql = "SELECT percentileState(upstream_response_time, 0.95) AS p95 " +"FROM log_main " +"WHERE timestamp BETWEEN ? AND ?";return jdbcTemplate.queryForList(sql, startTime, endTime);}
}

（7）、Controller 层（LogController.java）

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.util.List;
import java.util.Map;@RestController
public class LogController {@Autowiredprivate LogService logService;@GetMapping("/error-rate")public List<Map<String, Object>> getErrorRate() {return logService.getErrorRate();}@GetMapping("/response-time")public List<Map<String, Object>> getResponseTime(@RequestParam String startTime,@RequestParam String endTime) {return logService.getResponseTimeStats(startTime, endTime);}
}