日志之ClickHouse部署及替换ELK中的Elasticsearch

文章目录

1 ELK替换
- 1.1 Elasticsearch vs ClickHouse
- 1.2 环境部署
- - 1.2.1 zookeeper 集群部署
  - 1.2.2 Kafka 集群部署
  - 1.2.3 FileBeat 部署
  - 1.2.4 clickhouse 部署
  - - 1.2.4.1 准备步骤
    - 1.2.4.2 添加官方存储库
    - 1.2.4.3 部署&启动&连接
    - 1.2.4.5 基本配置服务
    - 1.2.4.6 测试创建数据库和表
    - 1.2.4.7 部署遇到问题
    - - 1.2.4.7.1 clikhouse 客户端无法查询 kafka 引擎表
      - 1.2.4.7.2 clickhouse 创建本地节点表，无法开启本地表 macro
      - 1.2.4.7.3 clickhouse 中节点数据已经存在
      - 1.2.4.7.4 分布式集群表无法查询
    - 1.2.4.8 clickhouse其他表
    - - 1.2.4.8.1 分布式表
      - 1.2.4.8.2 物化视图

1 ELK替换

点击了解日志之ELK使用讲解

1.1 Elasticsearch vs ClickHouse

ClickHouse 是一款高性能列式分布式数据库管理系统，对 ClickHouse 进行了测试，发现有下列优势：

ClickHouse 写入吞吐量大
单服务器日志写入量在 50MB 到 200MB/s，每秒写入超过 60w 记录数，是 ES 的 5 倍以上。
在 ES 中比较常见的写 Rejected 导致数据丢失、写入延迟等问题，在 ClickHouse 中不容易发生。
查询速度快
官方宣称数据在 pagecache 中，单服务器查询速率大约在 2-30GB/s；没在 pagecache 的情况下，查询速度取决于磁盘的读取速率和数据的压缩率。经测试 ClickHouse 的查询速度比 ES 快 5-30 倍以上。
ClickHouse 比 ES 服务器成本更低
一方面 ClickHouse 的数据压缩比比 ES 高，相同数据占用的磁盘空间只有 ES 的 1/3 到 1/30，节省了磁盘空间的同时，也能有效的减少磁盘 IO，这也是ClickHouse查询效率更高的原因之一。

另一方面 ClickHouse 比 ES 占用更少的内存，消耗更少的 CPU 资源。预估用 ClickHouse 处理日志可以将服务器成本降低一半。

支持功能\开源项目	ElasticSearch	ClickHouse
查询	java	c++
存储类型	文档存储	列式数据库
分布式支持	分片和副本都支持	分片和副本都支持
扩展性	高	低
写入速度	慢	快
CPU/内存占用	高	低
存储占用（54G日志数据导入）	高 94G(174%)	低 23G（42.6%）
精确匹配查询速度	一般	快
模糊匹配查询速度	快	慢
权限管理	支持	支持
查询难度	低	高
可视化支持	高	低
使用案例	很多	携程
维护难度	低	高

成本分析
在没有任何折扣的情况下，基于 aliyun 分析

成本项	标准	费用	说明	总费用
zookeeper 集群	2核4g 共享计算型 n4 50G SSD 云盘	222/月	3台高可用	666/月
kafka 集群	4核 8g 共享标准型 s650G SSD 云盘300G 数据盘	590/月	3台高可用	1770/月
filebeat 部署			混部相关的应用，会产生一定的内存以及磁盘开销，对应用的可用性会造成一定的影响。
clickhouse	16核32g 共享计算型 n450G SSD 云盘1000G 数据盘	2652/月	2台高可用	5304/月
总费用				7740/月

1.2 环境部署

1.2.1 zookeeper 集群部署

点击了解 zookeeper之Linux环境下的安装

yum install java-1.8.0-openjdk-devel.x86_64
/etc/profile 配置环境变量
更新系统时间
yum install  ntpdate
ntpdate asia.pool.ntp.org

mkdir zookeeper
mkdir ./zookeeper/data
mkdir ./zookeeper/logs
wget  --no-check-certificate https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gz
tar -zvxf apache-zookeeper-3.7.1-bin.tar.gz -C /usr/zookeeper

export ZOOKEEPER_HOME=/usr/zookeeper/apache-zookeeper-3.7.1-bin
export PATH=$ZOOKEEPER_HOME/bin:$PATH

进入ZooKeeper配置目录
cd$ZOOKEEPER_HOME/conf

新建配置文件
vi zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/usr/zookeeper/data
dataLogDir=/usr/zookeeper/logs
clientPort=2181
server.1=zk1:2888:3888
server.2=zk2:2888:3888
server.3=zk3:2888:3888

在每台服务器上执行，给zookeeper创建myid
echo"1" > /usr/zookeeper/data/myid
echo"2" > /usr/zookeeper/data/myid
echo"3" > /usr/zookeeper/data/myid

进入ZooKeeper bin目录
cd$ZOOKEEPER_HOME/bin
sh zkServer.sh start

1.2.2 Kafka 集群部署

点击了解 Kafka安装以及验证

mkdir -p /usr/kafka
chmod 777 -R /usr/kafka
wget  --no-check-certificate https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/3.2.0/kafka_2.12-3.2.0.tgz
tar -zvxf kafka_2.12-3.2.0.tgz -C /usr/kafka

不同的broker Id 设置不一样，比如 1,2,3
broker.id=1
listeners=PLAINTEXT://ip:9092
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dir=/usr/kafka/logs
num.partitions=5
num.recovery.threads.per.data.dir=3
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=3
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
zookeeper.connection.timeout.ms=30000
group.initial.rebalance.delay.ms=0

后台常驻进程启动kafka
nohup /usr/kafka/kafka_2.12-3.2.0/bin/kafka-server-start.sh /usr/kafka/kafka_2.12-3.2.0/config/server.properties   >/usr/kafka/logs/kafka.log >&1 &

/usr/kafka/kafka_2.12-3.2.0/bin/kafka-server-stop.sh

$KAFKA_HOME/bin/kafka-topics.sh --list --bootstrap-server  ip:9092

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server ip:9092 --topic test --from-beginning

$KAFKA_HOME/bin/kafka-topics.sh  --create --bootstrap-server  ip:9092  --replication-factor 2 --partitions 3 --topic xxx_data

1.2.3 FileBeat 部署

sudo rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch
Create a file with a .repo extension (for example, elastic.repo) in your /etc/yum.repos.d/ directory and add the following lines:
在/etc/yum.repos.d/ 目录下创建elastic.repo

[elastic-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

yum install filebeat
systemctl enable filebeat
chkconfig --add filebeat

FileBeat 配置文件说明，坑点 1（需设置 keys_under_root: true）。如果不设置kafka 的消息字段如下：

文件目录：/etc/filebeat/filebeat.yml

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /root/logs/xxx/inner/*.log
  json:  
如果不设置该索性，所有的数据都存储在message里面，这样设置以后数据会平铺。
       keys_under_root: true
output.kafka:
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]
  topic: 'xxx_data_clickhouse'
  partition.round_robin:
            reachable_only: false
            required_acks: 1
            compression: gzip
processors: 
剔除filebeat 无效的字段数据
    - drop_fields:  
        fields: ["input", "agent", "ecs", "log", "metadata", "timestamp"]
        ignore_missing: false

nohup ./filebeat -e -c /etc/filebeat/filebeat.yml > /user/filebeat/filebeat.log & 
输出到filebeat.log文件中，方便排查

1.2.4 clickhouse 部署

1.2.4.1 准备步骤

ClickHouse 官方 RPM 包通常针对 x86_64 架构编译，并依赖 SSE 4.2 指令集，检查当前CPU是否支持SSE 4.2，如果不支持，需要通过源代码编译构建

grep -q sse4_2 /proc/cpuinfo && echo"SSE 4.2 supported" || echo"SSE 4.2 not supported"
返回 "SSE 4.2 supported" 表示支持，返回 "SSE 4.2 not supported" 表示不支持

创建数据保存目录，将它创建到大容量磁盘挂载的路径

mkdir -p /data/clickhouse
修改/etc/hosts文件，添加clickhouse节点
举例：
10.190.85.92 bigdata-clickhouse-01
10.190.85.93 bigdata-clickhouse-02

服务器性能参数设置

cpu频率调节，将CPU频率固定工作在其支持的最高运行频率上，而不动态调节，性能最好
echo'performance' | tee /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor
内存调节，不要禁用 overcommit
echo 0 | tee /proc/sys/vm/overcommit_memory
始终禁用透明大页(transparent huge pages)，它会干扰内存分配器，从而导致显着的性能下降
echo 'never' | tee /sys/kernel/mm/transparent_hugepage/enabled

1.2.4.2 添加官方存储库

# 安装 yum-utils（用于管理 YUM 仓库）
yum install yum-utils 

# 导入 ClickHouse 的 GPG 密钥（用于验证包的真实性）
rpm --import https://repo.clickhouse.com/CLICKHOUSE-KEY.GPG

#运行命令后，可以检查是否成功导入密钥
# 如果成功导入，你会看到类似 gpg-pubkey-... --> GPG key for ClickHouse 的输出
rpm -q gpg-pubkey --qf '%{name}-%{version}-%{release} --> %{summary}\n'

# 添加 ClickHouse 仓库
yum-config-manager --add-repo https://repo.clickhouse.com/rpm/stable/x86_64

1.2.4.3 部署&启动&连接

部署服务，ClickHouse 提供两个主要包：clickhouse-server（服务端）和 clickhouse-client（客户端）

# 查看clickhouse可安装的版本
yum list | grep clickhouse

# 安装 ClickHouse 服务端和客户端
yum -y install clickhouse-server clickhouse-client
# 如果要指定版本
# yum install -y clickhouse-server-23.4.2.1 clickhouse-client-23.4.2.1

# 检查 ClickHouse 版本以确认安装成功：
clickhouse-client --version
输出示例：
ClickHouse client version 24.3.2.1 (official build).

启动服务，ClickHouse 安装后会创建一个 systemd 服务，名为 clickhouse-server。

# 启动服务
systemctl start clickhouse-server

#设置开机自启：
systemctl enable clickhouse-server

#检查服务状态：
systemctl status clickhouse-server

#检查服务日志：
cat /var/log/clickhouse-server/clickhouse-server.log
常见问题可能是权限或配置错误

连接服务，使用 clickhouse-client 连接到服务端，测试是否正常工作

#本地连接
clickhouse-client
默认连接 localhost:9000，用户为 default，无密码。

#如果设置了密码，可以用：
clickhouse-client --password

运行简单查询
在客户端提示符下运行：
SELECT 1;

1.2.4.5 基本配置服务

ClickHouse 的配置文件默认位于 /etc/clickhouse-server/，主要文件为 config.xml 和 users.xml

修改监听地址（允许远程访问）：默认只监听本地（127.0.0.1）。若需远程访问，编辑 /etc/clickhouse-server/config.xml

vi /etc/clickhouse-server/config.xml

找到 <listen_host>，修改为：<listen_host>0.0.0.0</listen_host>
0.0.0.0 表示监听所有网络接口。

设置默认用户密码：编辑 /etc/clickhouse-server/users.xml

vi /etc/clickhouse-server/users.xml
找到 <default> 用户的配置，添加或修改密码：<password>your_password</password>

或者使用 SHA256 哈希密码：echo -n "your_password" | sha256sum
将哈希值填入：<password_sha256_hex>your_hashed_password</password_sha256_hex>

#重启服务以应用配置：
systemctl restart clickhouse-server

修改日志级别为information，默认是trace，修改/etc/clickhouse-server/config.xml配置文件
<level>information</level>

执行日志所在目录：

正常日志：/var/log/clickhouse-server/clickhouse-server.log
异常错误日志：/var/log/clickhouse-server/clickhouse-server.err.log

1.2.4.6 测试创建数据库和表

进入客户端：clickhouse-client

-- 创建数据库：
CREATE DATABASE test;
-- 使用数据库：
USE test;
-- 创建表：
CREATE TABLE visits (
    id UInt64,
    duration Float64,
    url String,
    created DateTime
) ENGINE = MergeTree()
PRIMARY KEY id
ORDER BY id;

-- 插入数据：
INSERT INTO visits (id, duration, url, created) VALUES (1, 10.5, 'example.com', now());

-- 查询数据：
SELECT * FROM visits;

1.2.4.7 部署遇到问题

clickhouse 部署过程中遇到的一些问题如下：

1.2.4.7.1 clikhouse 客户端无法查询 kafka 引擎表

Kafka 引擎的特性：

ClickHouse 的 Kafka 引擎表是一种流式引擎（stream-like engine），设计目的是从 Kafka 消费数据，并将其传递给其他表（通常是物化视图或普通表）进行存储和处理。
默认情况下，ClickHouse 禁止直接查询 Kafka 引擎表（SELECT 操作），因为：
- Kafka 引擎表本质上是一个数据流，每次查询都会触发从 Kafka 消费数据的操作，可能导致重复消费或不一致的结果。
- Kafka 引擎表不存储数据，数据仅在 Kafka 中，查询可能会消耗大量资源（例如从 Kafka 的最早偏移量开始读取）。
ClickHouse 引入了安全限制，防止用户意外执行可能导致性能问题或数据不一致的查询

CREATE TABLE default.kafka_clickhouse_inner_log ON CLUSTER clickhouse_cluster (
    log_uuid   String ,
    date_partition   UInt32 ,
    event_name   String ,
    activity_name   String ,
    activity_type   String ,
    activity_id   UInt16 
) ENGINE = Kafka SETTINGS
    kafka_broker_list = 'kafka1:9092,kafka2:9092,kafka3:9092',
    kafka_topic_list = 'data_clickhouse',
    kafka_group_name = 'clickhouse_xxx',
    kafka_format = 'JSONEachRow',
    kafka_row_delimiter = '\n',
    kafka_num_consumers = 1;

ON CLUSTER clickhouse_cluster 是 ClickHouse 的 DDL 语法，用于在集群的所有节点上同步执行 DDL 语句。
解决方案：
需要在clickhouse client 创建加上 --stream_like_engine_allow_direct_select 1

1.2.4.7.2 clickhouse 创建本地节点表，无法开启本地表 macro

Code: 62. DB::Exception: There was an error on [127.0.0.1:9000]: Code: 62. DB::Exception: No macro 'shard' in config while processing substitutions in '/clickhouse/tables/default/bi_inner_log_local/{shard}' at '50' or macro is not supported here. (SYNTAX_ERROR) (version 22.5.2.53 (official build)). (SYNTAX_ERROR) (version 22.5.2.53 (official build))

创建本地表（使用复制去重表引擎）

create table default.bi_inner_log_local ON CLUSTER clickhouse_cluster (
    log_uuid   String ,
    date_partition   UInt32 ,
    event_name   String ,
    activity_name   String ,
    credits_bring   Int16 ,
    activity_type   String ,
    activity_id   UInt16 
) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/default/bi_inner_log_local/{shard}','{replica}')
  PARTITION BY date_partition
  ORDER BY (event_name,date_partition,log_uuid)
   SETTINGS index_granularity = 8192;

解决方案：在不同的 clickhouse 节点上配置不同的 shard，每一个节点的 shard 名称不能一致。

<macros>
    <shard>01</shard>
    <replica>example01-01-1</replica>
</macros>

1.2.4.7.3 clickhouse 中节点数据已经存在

报错：

Code: 253. DB::Exception: There was an error on : Code: 253. DB::Exception: Replica /clickhouse/tables/default/bi_inner_log_local/01/replicas/example01-01-1 already exists. (REPLICA_IS_ALREADY_EXIST) (version 22.5.2.53 (official build)). (REPLICA_IS_ALREADY_EXIST) (version 22.5.2.53 (official build))

解决方案：进入 zookeeper 客户端删除相关节点，然后再重新创建 ReplicatedReplacingMergeTree 表。这样可以保障每一个 clickhouse 节点都会去消费 kafka partition 的数据。

1.2.4.7.4 分布式集群表无法查询

报错：

Code: 516. DB::Exception: Received from 127.0.0.1:9000. DB::Exception: default: Authentication failed: password is incorrect or there is no user with such name. (AUTHENTICATION_FAILED) (version 22.5.2.53 (official build))

解决方案：

 <!--分布式表配置-->
<remote_servers>
	<clickhouse_cluster><!--集群名称, 可以自定义, 后面在新建库、表的时候需要用到集群名称-->
    	<shard>
	    	   <!--内部复制(默认false), 开启后, 在分布式表引擎下, 数据写入时-->
	           <!--每个分片只会去寻找一个节点写, 并不是每个都写-->
               <internal_replication>true</internal_replication>
               <replica>
                  <host>ip1</host>
                  <port>9000</port>
                   <user>default</user>
                   <password>xxxx</password>
              </replica>
         </shard>
         <shard>
                <internal_replication>true</internal_replication>
                <replica>
                    <host>ip2</host>
                    <port>9000</port>
                    <user>default</user>
                    <password>xxxx</password>
                </replica>
    	</shard>
	</clickhouse_cluster>
</remote_servers>

1.2.4.8 clickhouse其他表

1.2.4.8.1 分布式表

创建分布式表（根据 log_uuid 对数据进行分发，相同的 log_uuid 会发送到同一个 shard 分片上，用于后续合并时的数据去重）：

CREATE TABLE default.bi_inner_log_all ON CLUSTER clickhouse_cluster 
AS default.bi_inner_log_local
ENGINE = Distributed(clickhouse_cluster, default, bi_inner_log_local, xxHash32(log_uuid));

AS default.bi_inner_log_local：表示新表 default.bi_inner_log_all 的表结构（列定义）与已有表 default.bi_inner_log_local 相同。复制 default.bi_inner_log_local 的列定义（列名、类型等），但不复制数据。

ENGINE = Distributed(clickhouse_cluster, default, bi_inner_log_local, xxHash32(log_uuid))：指定表的引擎为 Distributed，表示这是一个分布式表。
Distributed 引擎的参数：

clickhouse_cluster：集群名称，指定查询分发的目标集群（与 ON CLUSTER 的集群名一致）。
default：本地表的数据库名，表示目标本地表位于 default 数据库。
bi_inner_log_local：本地表名，表示分布式表会将查询分发到名为 bi_inner_log_local 的本地表。
xxHash32(log_uuid)：分片键（sharding key），用于决定数据分发到哪个分片。
xxHash32 是一个哈希函数，对 log_uuid 列计算哈希值。
ClickHouse 根据哈希值将数据或查询分发到不同的分片，确保负载均衡。

Distributed 引擎的作用

不存储数据：
分布式表（default.bi_inner_log_all）本身不存储数据，仅作为一个逻辑层。
它将查询分发到集群中所有节点的本地表（default.bi_inner_log_local），并汇总结果。
查询分发：
当查询分布式表时，例如：SELECT * FROM default.bi_inner_log_all WHERE date_partition = 20230101;
ClickHouse 会将查询分发到集群中所有节点的 default.bi_inner_log_local 表。
每个节点独立执行查询，返回结果。
当前节点（发起查询的节点）汇总所有节点的结果，返回给客户端。
数据写入：
如果向分布式表插入数据，例如：INSERT INTO default.bi_inner_log_all VALUES (...);
ClickHouse 会根据分片键（xxHash32(log_uuid)）计算哈希值，将数据分发到对应的分片（节点）的 default.bi_inner_log_local 表。
分片键 xxHash32(log_uuid)
分片键决定数据如何分发到集群中的分片。
xxHash32(log_uuid) 对 log_uuid 列计算 32 位哈希值，ClickHouse 根据哈希值决定数据存储在哪个分片。

1.2.4.8.2 物化视图

创建物化视图，把 Kafka 消费表消费的数据同步到 ClickHouse 分布式表。

CREATE MATERIALIZED VIEW default.view_bi_inner_log ON CLUSTER clickhouse_cluster TO default.bi_inner_log_all AS 
SELECT 
    log_uuid ,
date_partition ,
event_name ,
activity_name ,
credits_bring ,
activity_type ,
activity_id 
FROM default.kafka_clickhouse_inner_log;

TO default.bi_inner_log_all：

TO 是物化视图的语法，用于指定目标表（target table）。
default.bi_inner_log_all 是目标表的名称，表示物化视图会将查询结果插入到 default 数据库中的 bi_inner_log_all 表。
作用：
- 物化视图会自动监听源表（default.kafka_clickhouse_inner_log）的新数据。
- 每次有新数据插入到源表时，物化视图会执行 SELECT 查询，并将结果插入到目标表 default.bi_inner_log_all

AS：