大数据技术

单机上部署kafka3.5，并配置sasl认证

Fri, 29 Mar 2024 19:33:24 +0800

概述

今天我要跟大家分享一下如何在单机上部署kafka3.5，并配置sasl认证（plaintext模式）。kafka是一个分布式的消息队列，可以用来处理大量的数据流。sasl是一种安全认证机制，可以保证kafka的通信安全。下面我会分几个步骤来介绍这个过程。

第一步，安装kafka

下载并解压kafka3.5的安装包。你可以从官网或者其他镜像站点下载kafka的压缩文件，然后解压到你想要的目录。例如，我把它解压到了/home/kafka目录。

第二步，修改kafka的配置文件

你需要修改两个文件，一个是server.properties，另一个是kafka_server_jaas.conf。server.properties是kafka的主要配置文件，里面有很多参数可以调整。kafka_server_jaas.conf是sasl认证的配置文件，里面定义了认证的方式和用户信息。这两个文件都在kafka的config目录下。

在server.properties中，你需要修改以下几个参数：

broker.id：这是kafka的唯一标识，每个broker都需要有一个不同的id。因为我们只有一个broker，所以可以设置为0。

listeners：这是kafka的监听地址和端口，可以有多个。我们需要添加一个sasl_plaintext类型的监听器，表示使用sasl和明文传输。例如，我们可以设置为listeners=PLAINTEXT://:9092,SASL_PLAINTEXT://:9093。

advertised.listeners：这是kafka对外公布的监听地址和端口，客户端会根据这个来连接kafka。我们需要和listeners保持一致，例如advertised.listeners=PLAINTEXT://:9092,SASL_PLAINTEXT://:9093。

zookeeper.connect：这是kafka连接zookeeper的地址和端口，zookeeper是一个分布式协调服务，用来管理kafka的元数据。因为我们只有一个zookeeper，所以可以设置为localhost:2181。

sasl.enabled.mechanisms：这是kafka启用的sasl认证方式，可以有多个。我们使用最简单的PLAIN方式，所以设置为sasl.enabled.mechanisms=PLAIN。

sasl.mechanism.inter.broker.protocol：这是kafka之间通信使用的sasl认证方式，必须是sasl.enabled.mechanisms中的一个。我们也设置为PLAIN。

authorizer.class.name：这是kafka使用的授权类，用来控制客户端对主题和消费组的访问权限。我们使用默认的SimpleAclAuthorizer，所以设置为authorizer.class.name=kafka.security.authorizer.AclAuthorizer。

super.users：这是kafka的超级用户列表，他们可以访问所有的主题和消费组，不受授权限制。我们可以设置为自己的用户名，例如super.users=User:admin。

在kafka_server_jaas.conf中，你需要添加以下内容：

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username=“admin”
password=“admin”
user_admin=“admin”
user_alice=“alice”;
};

这里我们定义了一个KafkaServer段，表示kafka服务器端使用的认证模块。我们使用PlainLoginModule，并指定了用户名和密码。我们还定义了两个用户，admin和alice，并分别给他们设置了密码。

第三步，启动zookeeper和kafka

你需要先启动zookeeper，然后启动kafka。你可以使用kafka自带的脚本来启动它们。例如，在/home/kafka目录下执行以下命令：

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

如果没有报错，那么恭喜你，你已经成功部署了kafka3.5，并配置了sasl认证。

第四步，测试kafka的功能和安全性

你可以使用kafka自带的命令行工具来创建主题，生产和消费消息，以及查看授权信息。例如，在/home/kafka目录下执行以下命令：

# 创建一个名为test的主题，有一个分区和一个副本
bin/kafka-topics.sh --create --topic test --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092

# 查看主题的信息
bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092

# 生产一些消息到test主题，使用sasl_plaintext协议和admin用户
bin/kafka-console-producer.sh --topic test --broker-list localhost:9093 --producer.config config/producer.properties

# 消费test主题的消息，使用sasl_plaintext协议和alice用户
bin/kafka-console-consumer.sh --topic test --bootstrap-server localhost:9093 --consumer.config config/consumer.properties

# 查看test主题的授权信息
bin/kafka-acls.sh --describe --topic test --bootstrap-server localhost:9092

在config目录下，你需要创建两个文件，producer.properties和consumer.properties，分别用来配置生产者和消费者的认证信息。例如，producer.properties的内容如下：

sasl.mechanism=PLAIN
security.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="admin" \
password="admin";

consumer.properties的内容如下：

sasl.mechanism=PLAIN
security.protocol=SASL_PLAINTEXT
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
username="alice" \
password="alice";

通过这些命令，你可以验证kafka的基本功能和sasl认证的效果。你可以尝试使用不同的用户和密码，或者不使用sasl_plaintext协议，看看会发生什么。

以上就是我今天要分享的内容，希望对你有帮助。如果你有任何问题或建议，欢迎留言。谢谢！

实战指南-构建安全可靠的Kafka集群并配置SASL/PLAIN认证

Fri, 29 Mar 2024 19:29:17 +0800

引言近年来，随着大数据和分布式系统的快速发展，数据传输和处理的需求也越来越大。在这个背景下，Apache Kafka作为一款高性能、高可靠性的分布式流媒体平台，吸引了众多企业和开发者的关注。与此同时，随着安全和数据保护意识的提升，数据传输的安全性也变得尤为重要。正因如此，深入了解Kafka集群的安装和认证机制显得尤为重要。本文将为大家介绍如何搭建一个具有SASL认证机制的Kafka集群环境，并分享一些实验经验和注意事项。

本文主要以3虚拟机测试环境安装了JDK 11, Zookeeper 3.8.3和Kafka 2.7.2集群，同时演示怎么来通过配置简单SASL认证后，怎么通过客户端来SASL密码方式登录Kafka服务器集群。在实践操作之前对Kafka、Kafka认证方式及SASL认证进行了简单的介绍。然而，仅仅通过用户名和密码进行连接认证是远远不够的，在实际应用中还是根据具体需求而选择不同的Kafka认证方式。

1 Kafka认证

关于Kafka

Apache Kafka是一种分布式流处理平台，它具有高吞吐量、扩展性和可靠性的特点。作为一个强大的分布式消息传递系统，Kafka被广泛应用于各种场景，如实时数据流处理、日志收集、事件驱动架构等。它通过高吞吐量、可靠性和可扩展性，为企业提供了处理大规模实时数据流的解决方案，满足了现代数据处理的需求。

随着数据传输和存储的增加，保护数据安全和隐私成为组织和个人的首要任务。Kafka提供了多种安全功能来确保数据的机密性、完整性和可用性。

关于Kafka认证

Kafka提供了多种认证方式来保护集群的安全性，包括以下几种常见的认证方式：

SSL/TLS认证：Kafka支持使用SSL/TLS协议对网络通信进行加密和认证。通过使用SSL/TLS证书对客户端和服务器进行身份验证，可以确保通信的机密性和完整性。
SASL/PLAIN认证：SASL/PLAIN是一种基于用户名密码的认证机制，可用于Kafka和其他应用程序之间的认证。它要求客户端提供用户名和密码来进行身份验证，并使用安全的方式进行传输。
SASL/GSSAPI(Kerberos)认证：SASL/GSSAPI认证是一种基于Kerberos的单点登录认证机制。它使用Kerberos票据来进行身份验证，通过集成现有的企业身份验证系统，实现了强大的安全性和无缝的用户体验。
自定义认证插件：Kafka还提供了自定义认证插件的扩展机制，允许用户根据特定需求实现自己的认证机制。通过自定义认证插件，可以与现有的身份验证系统或者第三方认证服务进行集成。

Kafka的认证方式的实现原理主要涉及以下几个步骤：

配置设置：通过Kafka的配置文件，设置认证方式和相关参数，如SSL证书的路径、用户名密码的配置等。
证书生成和配置：对于SSL/TLS认证，需要生成证书和私钥，并将其配置到Kafka的服务器端和客户端。证书用于对通信进行加密和认证。
用户认证信息配置：对于SASL/PLAIN和SASL/GSSAPI认证，需要在Kafka的配置文件中配置用户的认证信息，如用户名和密码，或者Kerberos票据。
服务器端验证：服务器端接收到客户端的连接请求后，根据配置的认证方式进行验证。对于SSL/TLS认证，服务器会验证客户端的证书；对于SASL/PLAIN和SASL/GSSAPI认证，服务器会验证客户端提供的用户名和密码或Kerberos票据。
客户端验证：客户端与服务器建立连接后，根据配置的认证方式提供相应的证书、用户名密码或Kerberos票据，进行身份验证。

通过以上步骤，Kafka能够实现不同的认证方式来保护集群的安全性。根据具体的需求和环境，选择合适的认证方式并正确配置，可以确保Kafka集群的安全访问和数据传输。

关于，

GNU SASL是简单身份验证和安全层框架以及一些常见SASL机制的实现。SASL被网络服务器(例如IMAP、SMTP、XMPP)用来请求来自客户端的身份验证，并在客户端中用来对服务器进行身份验证。

SASL是一个挑战-响应框架。在这里，服务器向客户机发出请求，客户机根据该请求发送响应。质询和响应是任意长度的字节数组，因此可以携带任何特定于机制的数据。工作简单原理图如下：

关于Kafka SASL认证

Kafka使用Java认证和授权服务(JAAS)进行SASL配置。

Kafka代理的JAAS配置： KafkaServer部分是每个KafkaServer/Broker使用的JAAS文件中的节名。为代理提供SASL配置选项，包括代理为进行代理间通信而建立的任何SASL客户机连接。如果多个侦听器配置为使用SASL，则可以使用侦听器名称的小写前缀加上句号，例如sasl_ssl.KafkaServer。 Client部分用于验证与zookeeper的SASL连接。它还允许代理在zookeeper节点上设置SASL ACL，从而锁定这些节点，以便只有代理可以修改它。有必要在所有代理中使用相同的主体名称。如果你想使用一个节名而不是Client，设置系统属性zookeeper.sasl.clientconfig为合适的名称(例如，-Dzookeeper.sasl.clientconfig=ZkClient)。
Kafka客户端的JAAS配置：客户端可以使用客户端配置属性sasl.jaas.config或使用类似于代理的静态JAAS配置文件来配置JAAS。

本文中后续以一个简单的例子较直观来解析kafka服务器端和客户端的配置连接。

2 安装Kafka集群

接下来例子以3个虚拟机Linux环境下，Kafka版本选择的是2.7的版本，在安装Kafka软件分别在3台机器上安装JDK和Zookeeper并配置zk正确运行下再安装Kafka。该环境下三台机器和IP地址分别如下：

VM65201 192.3.65.201
VM65202 192.3.65.202
VM65203 192.3.65.203

准备工作

下载所需的软件包：

JDK 11： https://www.oracle.com/java/technologies/javase/jdk11-archive-downloads.html 。选项“Linux x64 Compressed Archive”处进行 download

jdk-11.0.20_linux-x64_bin.tar.gz

Zookeeper 3.8.3：https://zookeeper.apache.org/releases.html

apache-zookeeper-3.8.3-bin.tar.gz

Kafka 2.7：官网下载 https://kafka.apache.org/downloads

kafka_2.13-2.7.2.tgz

安装Java（三台机器均需要执行）

--root下运行cd /usr/local
ls -lh jdk-11.0.20_linux-x64_bin.tar.gz
tar -xf jdk-11.0.20_linux-x64_bin.tar.gz 
cat >> /etc/profile <<EOFexport JAVA_HOME=/usr/local/jdk-11.0.20export PATH=\$JAVA_HOME/bin:\$PATHexport CLASSPATH=.:\$JAVA_HOME/lib:\$CLASSPATHEOF--确认java安装状况# source /etc/profile# java -versionjava version "11.0.20" 2023-07-18 LTS
Java(TM) SE Runtime Environment 18.9 (build 11.0.20+9-LTS-256)Java HotSpot(TM) 64-Bit Server VM 18.9 (build 11.0.20+9-LTS-256, mixed mode)

安装和配置Zookeeper集群

在三台机器上相同操作分别安装zookeeper软件，上传安装包到/usr/local，此处将软件安装到/usr/local下。

--root用户执行cd /usr/local/
ls -lh apache-zookeeper-3.8.3-bin.tar.gz
tar -xf apache-zookeeper-3.8.3-bin.tar.gz 
ls -l apache-zookeeper-3.8.3-bin/
--准备日志和数据目录
mkdir -p apache-zookeeper-3.8.3-bin/data apache-zookeeper-3.8.3-bin/logs

三台机器上相同操作分别配置zoo.cfg

cat > apache-zookeeper-3.8.3-bin/conf/zoo.cfg <<EOF
tickTime=2000
initLimit=10
syncLimit=5
dataLogDir=/usr/local/apache-zookeeper-3.8.3-bin/logs
dataDir=/usr/local/apache-zookeeper-3.8.3-bin/data
clientPort=2181
server.1=192.3.65.201:2888:3888
server.2=192.3.65.202:2888:3888
server.3=192.3.65.203:2888:3888
EOF

三台机器分别配置不同的myid

VM65201:/usr/local # echo "1" > /usr/local/apache-zookeeper-3.8.3-bin/data/myid
VM65202:/usr/local # echo "2" > /usr/local/apache-zookeeper-3.8.3-bin/data/myid
VM65203:/usr/local # echo "3" > /usr/local/apache-zookeeper-3.8.3-bin/data/myid

启动zookeeer集群

# /usr/local/apache-zookeeper-3.8.3-bin/bin/zkServer.sh startZooKeeper JMX enabled by default
Using config: /usr/local/apache-zookeeper-3.8.3-bin/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

--三台机器上分别查看状态
VM65201:/usr/local # /usr/local/apache-zookeeper-3.8.3-bin/bin/zkServer.sh statusZooKeeper JMX enabled by default
Using config: /usr/local/apache-zookeeper-3.8.3-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
VM65202:/usr/local # /usr/local/apache-zookeeper-3.8.3-bin/bin/zkServer.sh statusZooKeeper JMX enabled by default
Using config: /usr/local/apache-zookeeper-3.8.3-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower
VM65203:/usr/local # /usr/local/apache-zookeeper-3.8.3-bin/bin/zkServer.sh statusZooKeeper JMX enabled by default
Using config: /usr/local/apache-zookeeper-3.8.3-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leader

安装和配置Kafka集群

在三台机器上相同操作分别安装Kafka软件，上传安装包到/usr/local，此处将软件安装到/usr/local下。

--root用户下执行cd /usr/local
ls -lh kafka_2.13-2.7.2.tgz
tar -zxf kafka_2.13-2.7.2.tgz 
mkdir -p kafka_2.13-2.7.2/logs

三台机器上准备不同的配置文件。如果不配置listeners参数，只能本地localhost:9092端口访问。

# cp kafka_2.13-2.7.2/config/server.properties kafka_2.13-2.7.2/config/server.properties.org

--节点1
cat > kafka_2.13-2.7.2/config/server.properties <<EOF
broker.id=1
listeners=PLAINTEXT://192.3.65.201:9092
log.dirs=/usr/local/kafka_2.13-2.7.2/logs
zookeeper.connect=192.3.65.201:2181,192.3.65.202:2181,192.3.65.203:2181
EOF

--节点2
cat > kafka_2.13-2.7.2/config/server.properties <<EOF
broker.id=2
listeners=PLAINTEXT://192.3.65.202:9092
log.dirs=/usr/local/kafka_2.13-2.7.2/logs
zookeeper.connect=192.3.65.201:2181,192.3.65.202:2181,192.3.65.203:2181
EOF

--节点3
cat > kafka_2.13-2.7.2/config/server.properties <<EOF
broker.id=3
listeners=PLAINTEXT://192.3.65.203:9092
log.dirs=/usr/local/kafka_2.13-2.7.2/logs
zookeeper.connect=192.3.65.201:2181,192.3.65.202:2181,192.3.65.203:2181
EOF

启动Kafka集群

# kafka_2.13-2.7.2/bin/kafka-server-start.sh -daemon kafka_2.13-2.7.2/config/server.properties

--如果失败 可以去除daemon通过前台启动或者查看logs/下日志
# kafka_2.13-2.7.2/bin/kafka-server-start.sh kafka_2.13-2.7.2/config/server.properties
...

3 SASL/PLAIN认证

默认连接操作

通过Kafka软件下自带的工具连接 - 创建topic

# kafka_2.13-2.7.2/bin/kafka-topics.sh --list --bootstrap-server 192.3.65.201:9092,192.3.65.202:9092,192.3.65.203:9092
# kafka_2.13-2.7.2/bin/kafka-topics.sh --create  --bootstrap-server 192.3.65.201:9092,192.3.65.202:9092,192.3.65.203:9092 --replication-factor 3 --partitions 1 --topic test1
Created topic test1.
# kafka_2.13-2.7.2/bin/kafka-topics.sh --list --bootstrap-server 192.3.65.201:9092,192.3.65.202:9092,192.3.65.203:9092
test1

配置SASL/PLAIN认证

zookeeper上准备授权用户 admin，密码admin2023。

--其中一台机器上登录zk
VM65201:/usr/local # apache-zookeeper-3.8.3-bin/bin/zkCli.sh 
...
[zk: localhost:2181(CONNECTED) 0] create /digest user
Created /digest
[zk: localhost:2181(CONNECTED) 1] addauth digest admin:admin2023
[zk: localhost:2181(CONNECTED) 2] setAcl /digest auth:admin:admin2023:crwda
[zk: localhost:2181(CONNECTED) 3] quit

三台机器上分别修改配置

--root用户# cp kafka_2.13-2.7.2/config/server.properties kafka_2.13-2.7.2/config/server-sasl.properties# vi kafka_2.13-2.7.2/config/server-sasl.properties --修改listener中PLIANTEXT为SASL_PLAINTEXT#listeners=PLAINTEXT://192.3.65.201:9092listeners=SASL_PLAINTEXT://:9092

--增加如下内容指定认证方式
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

--指定超级管理员用户
super.users=User:admin

准备服务器端jaas文件，并指定用户名和密码。

--三台服务器均需执行
cat >>  kafka_2.13-2.7.2/config/kafka-server-jaas.conf <<EOF
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin2023"
    user_admin="admin2023";
};
EOF

重新启动kafka服务器，三台服务器均需执行。

--停止服务器# kafka_2.13-2.7.2/bin/kafka-server-stop.sh --修改启动脚本# cp kafka_2.13-2.7.2/bin/kafka-server-start.sh kafka_2.13-2.7.2/bin/kafka-server-start-sasl.sh # vi kafka_2.13-2.7.2/bin/kafka-server-start-sasl.sh exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"-->改为exec $base_dir/kafka-run-class.sh $EXTRA_ARGS -Djava.security.auth.login.config=/usr/local/kafka_2.13-2.7.2/config/kafka-server-jaas.conf kafka.Kafka "$@"--启动带SASL的服务器# kafka_2.13-2.7.2/bin/kafka-server-start-sasl.sh kafka_2.13-2.7.2/config/server-sasl.properties ...
--后台启动# kafka_2.13-2.7.2/bin/kafka-server-start-sasl.sh -daemon kafka_2.13-2.7.2/config/server-sasl.properties # jps23638 QuorumPeerMain27133 Kafka27151 Jps

另外一台有Kafka客户端的机器上配置Kafka客户端认证文件并登录集群

--由于登录Kafka认证会严重服务器主机名，所以需先配置/etc/hosts
# cat >>/etc/hosts <<EOF
192.3.65.201     VM65201
192.3.65.202     VM65202
192.3.65.203     VM65203
EOF

--不配置/etc/hosts会出现如下报错
[2023-11-08 09:37:58,961] WARN [Controller id=1, targetBrokerId=2] Error connecting to node VM65202:9092 (id: 2 rack: null) (org.apache.kafka.clients.NetworkClient)
java.net.UnknownHostException: VM65202: Name or service not known

--配置客户端配置
# cat > /tmp/jaas.conf <<EOF
KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin2023";
};
EOF
# cat > /tmp/security-client.properties <<EOF
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
EOF

# bin/kafka-topics.sh --list --bootstrap-server 192.3.65.201:9092,192.3.65.202:9092,192.3.65.203:9092 --command-config /tmp/security-client.properties 
test1
# bin/kafka-topics.sh --describe --bootstrap-server 192.3.65.201:9092,192.3.65.202:9092,192.3.65.203:9092 --command-config /tmp/security-client.properties --topic test1
Topic: test1    PartitionCount: 1       ReplicationFactor: 3    Configs: 
        Topic: test1    Partition: 0    Leader: 3       Replicas: 3,2,1 Isr: 3,1,2

--向topic生成数据
# bin/kafka-console-producer.sh  --bootstrap-server 192.3.65.201:9092,192.3.65.202:9092,192.3.65.203:9092 --topic test1 --producer-property security.protocol=SASL_PLAINTEXT --producer-property sasl.mechanism=PLAIN 
>abc
>xyz
>this is test

# bin/kafka-console-consumer.sh  --bootstrap-server 192.3.65.201:9092,192.3.65.202:9092,192.3.65.203:9092 --topic test1 --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN --from-beginning 
abc
xyz
this is test
>^C

通过SASL/PLAIN认证连接

解释SASL/PLAIN认证的概念和工作原理提供Kafka配置文件的修改示例，启用和配置SASL/PLAIN认证

总结

通过本文的实验和操作指导，我们成功搭建了一个具有SASL认证机制的Apache Kafka集群环境。我们首先安装了Java、Zookeeper和Kafka，然后配置了SASL认证，并进行了客户端和服务器的连接认证。在此过程中，我们深入理解了Kafka集群的安装和认证机制，并且获取了一些实践经验和注意事项。

参考文档

GUN SASL库：https://www.gnu.org/software/gsasl/
An Introduction to Java SASL: https://www.baeldung.com/java-sasl
Apache Kafka官网：http://kafka.apache.org/

Kafka的监听地址配置

Fri, 29 Mar 2024 14:24:29 +0800

本文使用版本：Kafka 3.6.0

有时我们会碰到网络是通畅的，但却连不上Kafka，特别是在多网卡环境或者云环境上很容易出现。这个其实和Kafka的监听配置有关系。本文介绍监听相关的配置。目前监听相关的参数主要有下面几个：

listeners
advertised.listeners
listener.security.protocol.map
inter.broker.listener.name
security.inter.broker.protocol
advertised.host.name （历史遗留，已废弃，勿使用）
advertised.port （历史遗留，已废弃，勿使用）
host.name （历史遗留，已废弃，勿使用）

其中最重要的就是 listeners 和 advertised.listeners ：集群启动时监听listeners配置的地址，并将 advertised.listeners 配置的地址写到Zookeeper里面，作为集群元数据的一部分。我们可以将客户端（生产者/消费者）连接Kafka集群进行操作的过程分成2步：

通过 listeners 配置的连接信息（ip/host）连接到某个Broker（broker会定期获取并缓存zk中的元数据信息），获取元数据中 advertised.listeners 配置的地址信息。

通过第1步获取的 advertised.listeners 连接信息和Kafka集群通信（读/写）。

所以在存在内外网隔离的虚拟化环境中（比如Docker、公有云），外部客户端经常会出现可以连接到Kafka（第1步），但发送/消费数据时报连接超时（第2步），就是因为 listeners 配置的是外网地址，而 advertised.listeners 配置的却是内网地址。那这几个参数该如何配置呢？

先看连接信息的配置格式： {listener名字}://{HOST/IP}:{PORT} 。HOST/IP、PORT很清楚，主要是这个“ listener名字 ”字段。要理解这个得了解 listener.security.protocol.map 这个配置项：它的用途是配置listener名字和协议的映射（所以它是一个key-value的map），key是“listener名字”，value是“协议名称”，其默认值是“listener名字”和“协议名称”一样。有点绕，举个例子，比如：PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL，冒号前面是key，即协议名字；后面是value，即协议名称。listener名字我们可以随便起，而协议名称则是固定可枚举的一个范围。所以如果我们自定义了listener名字，那就需要显式的设置其对应的协议名。

inter.broker.listener.name 和 security.inter.broker.protocol 都是用于配置Broker之间通信的，前者配置名称（即 listener.security.protocol.map 中的key），后者配置协议（即 listener.security.protocol.map 中的value），默认值是 PLAINTEXT 。这两个配置项同时只能配置一个。

为什么一个连接要搞这么复杂呢？主要是为了各种不同的场景需求。下面举一个复杂一点的应用场景进行说明。比如我们在一个公有云上面部署了一个Kafka集群，该环境有一个外网地址 external_hostname 和一个内网地址 internal_hostname ；且在内部中是无法获取外网地址的（公有云大多都是这样的）。然后想实现内部客户端访问集群时走内部地址，且不需要加密；而外部客户端访问时则走外部地址，且需要加密。要实现这个需求，可以对集群进行如下配置：

listener.security.protocol.map=INTERNAL:,PLAINTEXT,EXTERNAL:SSL
listeners=INTERNAL://0.0.0.0:19092,EXTERNAL://0.0.0.0:9092
advertised.listeners=INTERNAL://{internal_hostname}:19092,EXTERNAL://{external_hostname}:9092
inter.broker.listener.name=INTERNAL

其实更进一步，我们还可以通过可选的control.plane.listener.name参数单独定制集群Controller节点与其他Broker节点的连接，那配置信息就变为：

listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:SSL,CONTROL:SSL
listeners=INTERNAL://0.0.0.0:19092,EXTERNAL://0.0.0.0:9092
advertised.listeners=INTERNAL://{internal_hostname}:19092,EXTERNAL://{external_hostname}:9092,CONTROL://{control_ip}:9094
inter.broker.listener.name=INTERNAL
control.plane.listener.name=CONTROL

最后给出这些配置项的默认值和一些注意事项：

listeners 如果不显式的配置，那会监听所有网卡，相当于配置了0.0.0.0。该配置项里面listeners名字和端口都必须是唯一的，不能重复。
advertised.listeners 如果不配置，默认使用 listeners 配置的值。如果 listeners 也没有显式配置，则使用 java.net.InetAddress.getCanonicalHostName() 获取的IP地址。如果 listeners 配置的是0.0.0.0，则必须显式的配置 advertised.listeners ，因为这个配置项必须是一个具体的地址，不允许是0.0.0.0（因为客户端无法根据这个地址连接到Broker）。另外，advertised.listeners 中的端口允许重复。
对于 listeners 和 advertised.listeners ，有多个地址的时候，每一个地址都必须按照 {listener名字}://{HOST/IP}:{PORT} 格式进行配置，多个地址用英文逗号分隔。
如果集群所有节点的hostname在客户端和服务端各节点之间可以正确解析，优先使用hostname，而不是IP。因为代码里面使用了 java.net.InetAddress.getCanonicalHostName() ，有时使用IP会出现访问不通的情况。

总结： listeners 地址是用于首次连接的； advertised.listeners 的地址是会写到zk里面，客户端通过listeners地址建立连接获取该地址信息，然后通过该地址和集群交互。所以对于客户端，这2个地址必须都是可以访问的才可以。

KAFAK 配置内外网分流，实现同时支持内网，外网访问kafka

Wed, 27 Mar 2024 16:55:20 +0800

这里的内网和外网并不必须是是我们通常说的公司内部网络和公网，只要是两块网卡都可以，不管是这两块网卡是公网、内网、甚至是k8s的虚拟网络两个网段的主要目的是实现分流，一个负责节点间通信，一个负责对外收发消息

总结一句话：

listeners 指明 kafka 当前节点监听本机的哪个网卡
advertised.listeners 指明客户端通过哪个 ip 可以访问到当前节点

1. kafka的两个配置listeners和advertised.listeners

1.1 listeners:

kafka监听的网卡的ip，假设你机器上有两张网卡，内网192.168.0.213和外网101.89.163.1
如下配置
listeners=PLAINTEXT://192.168.0.213:9092

那么kafka只监听内网网卡，即只接收内网网卡的数据，如果如下配置：
listeners=PLAINTEXT://101.89.163.1:9092

那么kafka只监听外网网卡，即只接收外网网卡的数据。
当然ip可以配置成0.0.0.0，监听所有网卡。
总结：
listeners 解决的是 kafka 监听来自于哪个网卡的请求。
1.2 advertised.listeners
一般配置成这样：

listeners=PLAINTEXT://192.168.0.213:9092
advertised.listeners=PLAINTEXT://101.89.163.1:9092

配置过 zookeeper 的都知道， zookeeper 的配置文件中写明了所有 zookeeper 节点的 ip 地址，以便启动后节点之间的互相通信。我们观察 kafka 的配置文件 server.properties，会发现里面并没有记录其他 kafka 节点的地址，即各个节点都不知道其他节点的 ip 地址。那么 kafka 集群的各个节点之间是怎么通信的呢？

kafka 节点启动后，会向 zookeeper 注册自己，同时告诉 zookeeper 自身的通信地址，这个地址就是配置文件中的 advertised.listeners，如果没有配置 advertised.listeners，就会使用listeners。同时从 zookeeper 中获取兄弟节点的这个地址，以便与兄弟节点通信。即 kafka 节点是从 zookeeper 获取的其他节点的通信地址。

我们使用客户端以一个 ip 地址首次连接 kafka 节点后，节点返回给客户端的 kafka 集群地址就是从 zookeeper 中获得的这些地址，也就是各个节点配置的 advertised.listeners，包括当前连接的节点。所以可能客户端后续访问当前节点的 ip 地址有可能和首次连接的 ip 地址并不一样。

2. 三种情景，搭配使用这两个配置

2.1. 只需要内网访问kafka

这个很简单，只配置 listeners 并且使用内网 ip

listeners=PLAINTEXT://192.168.0.213:9092

客户端访问 kafka 节点，以及与其他节点通信都用这个 ip 地址。

2.2 需要外网访问

2.2.1 机器有外网网卡

最简单的一个方法，只配置 listeners 并且使用外网 ip

listeners=PLAINTEXT://101.89.163.1:9092

所有请求都走外网网卡，虽然有问题（所有流量走外网），但是能够正常使用。

2.2.2 机器没有外网网卡

如果机器是通过转发或者映射搞出来的外网ip（例如你访问腾讯新闻某个页面使用的 ip，这个 ip 和实际的网卡并不在为你提供首页内容的那台机器上），此时kafka无法监听这个外网ip，启动就会报错。
需要配置一下 advertised.listeners，例如：

listeners=PLAINTEXT://192.168.0.213:9092

advertised.listeners=PLAINTEXT://101.89.163.1:9092

此时 kafka 客户端访问 kafka 集群的一个完整流程是：

客户端访问 101.89.163.1:9092，该请求被发送到 kafka 节点宿主机的内网地址 192.168.0.213:9092 上（也可以是其他 kafka 节点，效果是一样的），第一次请求成功，此次请求申请获得 kafka 集群的所有节点的地址。
kafka 从 zookeeper 拿到自己和其他兄弟节点注册在 zookeeper 的 advertised.listeners，作为 kafka 集群各个节点的的服务 ip 返回给客户端。
客户端拿这些返回的服务 ip 访问kafka集群，该请求从实际拥有该 ip 的机器被转发到被具体 kafka 节点所在的内网ip，访问到了 kafka 节点，
当然有外网网卡时你也可以配置

listeners=PLAINTEXT://101.89.163.1:9092

advertised.listeners=PLAINTEXT://101.89.163.1:9092

但是这毫无意义。

2.2.3 为什么客户端需要在第一次请求中获取 kafka 各个节点的服务 ip

因为你可以在启动客户端时只配置一个 kafka 节点的地址而不是列出所有节点（这样是不推荐的），但是客户端必须具有访问集群中的每一个节点的能力（收消息、发消息都可能面向所有节点）。
对于 2.2 的情况，如果去掉

advertised.listeners=PLAINTEXT://101.89.163.1:9092

你会发现虽然你在启动 kafka 客户端时配置的访问地址是101.89.163.1:9092，但是 kafka 客户端启动时报错：

Connection to node -1[192.168.0.213:9092] could not be established. Broker may not be available.

为什么明明配置的是101.89.163.1:9092启动时连接的是192.168.0.213？这就是因为不配置advertised.listeners 则使用listeners 代替并注册到zookeeper 中，客户端拿到的 kafka 节点 ip 就是listeners配置的内网 ip 192.168.0.213。

3. 内外网分流

2中两种情况都有一个问题，所有请求包括 kafka 节点之间的通信都走的外网，这既不方便（外网访问的各种权限控制很麻烦），也不经济，更不安全。
这时候就要配置内外网。
配置1（针对没有实际的外网网卡，即2.2的情况）：

listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
listeners=INTERNAL://192.168.0.213:9092,EXTERNAL://192.168.0.213:19092
advertised.listeners=INTERNAL://192.168.0.213:9092,EXTERNAL://101.89.163.9:19092
inter.broker.listener.name=INTERNAL

配置2（针对有实际的外网网卡，即2.1的情况）：

listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
listeners=INTERNAL://192.168.0.213:9092,EXTERNAL://101.89.163.9:19092
advertised.listeners=INTERNAL://192.168.0.213:9092,EXTERNAL://101.89.163.9:19092
inter.broker.listener.name=INTERNAL

这两配置的区别是只有 listeners 的 EXTERNAL 使用的 ip 不一样，一个使用内网 ip ，一个使用外网 ip 。

对内网 INTERNAL 的配置就不说了，主要用于节点间通信，无论 listeners 还是 advertised.listeners 都配置成内网 ip，自然所有请求都走内网；主要是外网 EXTERNAL 配置：

1）如果机器的外网 ip 是映射来的，listeners 的 EXTERNAL 使用内网ip，客户端拿到的是 advertised.listeners 配置的外网 ip，对于该外网 ip 的请求会被转发到对应的内网 ip，即访问到了对应的 kafka 节点，请求完成闭环。不要问怎么转发的，既然机器被映射了外网 ip，自然能够转发，有问题问网管
2）如果你的kafka宿主机有外网网卡，listeners 的EXTERNAL 使用用外网ip，客户端拿到的是 advertised.listeners 配置的外网 ip，直接就访问到了对应的 kafka 节点。

clickhouse修改数据存储目录

Fri, 02 Feb 2024 19:20:53 +0800

clickhouse默认数据目录是： /var/lib/clickhouse。

修改方案：先停止数据库，然后转移该目录到指定位置，最后软连接回原地址并启动；即无需修改config.xml实现对数据目录的修改。

1、停止数据库

service clickhouse-server stop

mkdir /cc/soft/clickhouse

2、迁移数据目录

修改权限，否则没权限就启动不了

数据量太大，较慢，耐心等待。

mv /var/lib/clickhouse/* /cc/soft/clickhouse/

chown -R clickhouse:clickhouse /cc/soft/clickhouse

3、建立软连接

ln -s /cc/soft/clickhouse /var/lib/

#检查创建的软连接

ls -l /var/lib/

4、启动数据库-完成数据目录变更

第二种方式是在修改congfig配置文件

参考：修改clickhouse的默认存储路径_clickhouse日志路径-CSDN博客

clickhouse创建角色和用户

Tue, 30 Jan 2024 10:16:39 +0800

clickhouse版本：22.2.2.1

更改default账户权限

ClickHouse提供了一个default账号，这个账号有所有的权限，但是不能使用SQL驱动方式的访问权限和账户管理。我们需要在配置文件中修改default账户，使其能够通过SQL驱动方式添加角色和用户。默认default用户是没有办法创建用户的，这时候需要我们修改配置文件users.xml，便能创建用户了,详见官网

修改 /etc/clickhouse-server/users.xml 文件，在<users>:<default>标签中，将<access_management></access_management>中的值从0改成1。

<networks>
        <ip>::/0</ip>
    </networks>

    <!-- Settings profile for user. -->
    <profile>default</profile>

    <!-- Quota for user. -->
    <quota>default</quota>

    <!-- User can create other users and grant rights to them. -->
    <access_management>1</access_management> # 该部分控制权限，把原本注释去掉就能建用户了
    # 下面这几行也要加，不然授权all的时候会提示权限不足
    <named_collection_control>1</named_collection_control>
    <show_named_collections>1</show_named_collections>
    <show_named_collections_secrets>1</show_named_collections_secrets>
</default>
# 修改后需要重启clickhouse服务

2. 重启clickhouse-server

service clickhouse-server restart

创建管理员

创建管理员账户，并为其分配所有权限。

CREATE USER IF NOT EXISTS admin IDENTIFIED WITH sha256_password BY 'password'GRANT ALL ON *.* TO admin WITH GRANT OPTION

登录管理员账户

假设有一个数据库名为test。接下来，登录管理员账户，然后为test添加读写账户。

clickhouse-client -u admin --password password

创建普通角色

创建只读角色

该角色只能读取test的数据。

CREATE ROLE ro
GRANT SELECT ON test.* TO ro

创建读写角色

该角色可以读写test的数据。

CREATE ROLE rw
GRANT SELECT, INSERT, ALTER UPDATE, ALTER DELETE ON test.* TO rw

创建普通账户

我们分别创建账户reader和writer，然后把只读角色ro分配给reader，把读写角色分配给writer。这样，reader拥有只读权限，writer拥有读写权限。

CREATE USER IF NOT EXISTS reader IDENTIFIED WITH sha256_password BY 'password'
GRANT ro TO reader

CREATE USER IF NOT EXISTS writer IDENTIFIED WITH sha256_password BY 'password'
GRANT rw TO writer

参考资料：ClickHouse中创建用户、数据库并进行权限分配-易微帮 (ewbang.com)

ClickHouse核心引擎MergeTree解读

Thu, 11 Jan 2024 21:58:51 +0800

ClickHouse 是俄罗斯最大的搜索引擎Yandex在2016年开源的数据库管理系统（DBMS），主要用于联机分析处理（OLAP）。其采用了面向列的存储方式，性能远超传统面向行的DBMS，近几年受到广泛关注。

本文将介绍ClickHouse MergeTree系列表引擎的相关知识，并通过示例分析MergeTree存储引擎的数据存储结构。

1 MergeTree表引擎简介

MergeTree（合并树）系列表引擎是ClickHouse提供的最具特色的存储引擎。MergeTree引擎支持数据按主键、数据分区、数据副本以及数据采样等特性。官方提供了包括MergeTree、ReplacingMergeTree、SummingMergeTree、AggregatingMergeTree、CollapsingMergeTree、VersionedCollapsingMergeTree、GraphiteMergeTree等7种不同类型的MergeTree引擎的实现，以及与其相对应的支持数据副本的MergeTree引擎（Replicated*）。

首先来介绍一下MergeTree核心引擎：

ReplacingMergeTree：在后台数据合并期间，对具有相同排序键的数据进行去重操作。
SummingMergeTree：当合并数据时，会把具有相同主键的记录合并为一条记录。根据聚合字段设置，该字段的值为聚合后的汇总值，非聚合字段使用第一条记录的值，聚合字段类型必须为数值类型。
AggregatingMergeTree：在同一数据分区下，可以将具有相同主键的数据进行聚合。
CollapsingMergeTree：在同一数据分区下，对具有相同主键的数据进行折叠合并。
VersionedCollapsingMergeTree：基于CollapsingMergeTree引擎，增添了数据版本信息字段配置选项。在数据依据ORDER BY设置对数据进行排序的基础上，如果数据的版本信息列不在排序字段中，那么版本信息会被隐式的作为ORDER BY的最后一列从而影响数据排序。
GraphiteMergeTree：用来存储时序数据库Graphites的数据。

MergeTree是该系列引擎中最核心的引擎，其他引擎均以MergeTree为基础，并在数据合并过程中实现了不同的特性，从而构成了MergeTree表引擎家族。下面我们通过MergeTree来具体了解MergeTree表系列引擎。

2 MergeTree引擎

2.1 表创建

创建MergeTree的DDL如下所示：

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster] (
      name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1] [TTL expr1], 
      name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2] [TTL expr2], 
      ... ) ENGINE = MergeTree()
  ORDER BY expr 
  [PARTITION BY expr] 
  [PRIMARY KEY expr] 
  [SAMPLE BY expr] 
  [TTL expr [DELETE|TO DISK 'xxx'|TO VOLUME 'xxx'], ...] 
  [SETTINGS name=value, ...

这里说明一下MergeTree引擎的主要参数：

[必填选项]

ENGINE：引擎名字，MergeTree引擎无参数。
ORDER BY：排序键，可以由一列或多列组成，决定了数据以何种方式进行排序，例如ORDER BY（CounterID, EventDate）。如果没有显示指定PRIMARY KEY，那么将使用ORDER BY作为PRIMARY KEY。通常只指定ORDER BY即可。

[选填选项]

PARTITION BY：分区键，指明表中的数据以何种规则进行分区。分区是在一个表中通过指定的规则划分而成的逻辑数据集。分区可以按任意标准进行，如按月、按日或按事件类型。为了减少需要操作的数据，每个分区都是分开存储的。
PRIMARY KEY：主键，设置后会按照主键生成一级索引（primary.idx），数据会依据索引的设置进行排序，从而加速查询性能。默认情况下，PRIMARY KEY与ORDER BY设置相同，所以通常情况下直接使用ORDER BY设置来替代主键设置。
SAMPLE BY：数据采样设置，如果显示配置了该选项，那么主键配置中也应该包括此配置。例如 ORDER BY CounterID / EventDate / intHash32（UserID）、SAMPLE BY intHash32（UserID）。
TTL：数据存活时间，可以为某一字段列或者一整张表设置TTL，设置中必须包含Date或DateTime字段类型。如果设置在列上，那么会删除字段中过期的数据。如果设置的是表级的TTL，那么会删除表中过期的数据。如果设置了两种类型，那么按先到期的为准。例如，TTL createtime + INTERVAL 1 DAY，即一天后过期。使用场景包括定期删除数据，或者定期将数据进行归档。
index_granularity：索引间隔粒度。MergeTree索引为稀疏索引，每index_granularity个数据产生一条索引。index_granularity默认设置为8092。
enable_mixed_granularity_parts：是否启动index_granularity_bytes来控制索引粒度大小。
index_granularity_bytes：索引粒度，以字节为单位，默认10Mb。
merge_max_block_size：数据块合并最大记录个数，默认8192。
merge_with_ttl_timeout：合并频率最小时间间隔，默认1天。

2.2 数据存储结构

首先创建一个test表，DDL如下:

CREATE TABLE test.test 
( 
    id        UInt64, 
    type      UInt8, 
    create_time DateTime 
) ENGINE = MergeTree() 
  PARTITION BY toYYYYMMDD(create_time) 
  ORDER BY (id) 
  SETTINGS index_granularity = 4;

test表包括id、type、create等三个字段，其中以create_time日期字段作为分区键，并将日期格式转化为YYYYMMDD。按照id字段进行排序。由于没有显式设置主键，所以引擎默认使用ORDER BY设置的id列作为索引字段，并生成索引文件。index_granularity设置为4，意味着每4条数据产生一条索引数据。

插入一条测试数据：

insert into test.test(id, type, create_time) VALUES (1, 1, toDateTime('2021-03-01 00:00:00'));

使用如下命令查看test表分区相关信息：

 SELECT     
    database,     
    table,     
    partition,     
    partition_id,     
    name,     
    active,     
    path 
 FROM system.parts 
 WHERE table = 'test'

返回结果如下图所示：

从上图中可以看到test表中返回了一条partitionid为20210301的数据分区的记录，从name字段中我们可以得知，此分区的目录名为20210301_8_8_0。 20210301_8_8_0这个目录名字到底有什么含义呢？下面来介绍一下分区规则以及分区目录的命名规则。

2.2.1 数据分区ID生成规则

数据分区规则由分区ID决定，分区ID由PARTITION BY分区键决定。根据分区键字段类型，ID生成规则可分为：

未定义分区键：没有定义PARTITION BY，默认生成一个目录名为all的数据分区，所有数据均存放在all目录下。
整型分区键：分区键为整型，那么直接用该整型值的字符串形式做为分区ID。
日期类分区键：分区键为日期类型，或者可以转化成日期类型。
其他类型分区键：String、Float类型等，通过128位的Hash算法取其Hash值作为分区ID。

上面我们插入一条日期为2021-03-01 00:00:00的数据，对该字段格式化后生成的数据分区id就是20210301。

2.2.2 数据分区目录命名规则

目录命名规则如下：

PartitionId_MinBlockNum_MaxBlockNum_Level

PartitionID：分区id，例如20210301。
MinBlockNum：最小分区块编号，自增类型，从1开始向上递增。每产生一个新的目录分区就向上递增一个数字。
MaxBlockNum：最大分区块编号，新创建的分区MinBlockNum等于MaxBlockNum的编号。
Level：合并的层级，被合并的次数。合并次数越多，层级值越大。

从上图可知，此分区的分区id为20210301，当前分区的MinBlockNum和MinBlockNum均为8，而level为0，表示此分区没有合并过。

2.3 数据分区文件组织结构

在了解了分区目录名字的生成规则后，下面来看看数据分区目录下的文件组织结构。以2021030188_0分区为例：

从图中可以看到，目录中的文件主要包括bin文件、mrk文件、primary.idx文件以及其他相关文件。

bin文件：数据文件，存储的是某一列的数据。数据表中的每一列都对应一个与其字段名相同的bin文件，例如id.bin存储的是表test中id列的数据。
mrk文件：标记文件，每一列都对应一个与其字段名相同的标记文件，标记文件在idx索引文件和bin数据文件之间起到了桥梁作用。以mrk2结尾的文件，表示该表启用了自适应索引间隔。
primary.idx文件：主键索引文件，用于加快查询效率。
count.txt：数据分区中数据总记录数。上述20210301_8_8_0的数据分区中，该文件中的记录总数为1。
columns.txt：表中所有列数的信息，包括字段名和字段类型。
partion.dat：用于保存分区表达式的值。上述20210301_8_8_0的数据分区中该文件中的值为20210301。
minmax_create_time.idx：分区键的最大最小值。
checksums.txt：校验文件，用于校验各个文件的正确性。存放各个文件的size以及hash值。

2.3.1 数据文件

MergeTree中，每列都对应一个bin文件单独存放该列数据。例如，id.bin存放的是id列的数据。所有数据都经过数据压缩、排序，最后以数据块的形式写入bin文件中。bin中数据以压缩数据块为单位写入文件中。每个数据块由头信息和压缩数据组成。头部信息包括校验和、数据压缩算法、数据压缩前大小和压缩后大小组成。压缩数据由granule组成，granule大小与index_granularity相关。

2.3.2 索引文件

MergeTree索引为稀疏索引，它并不索引单条数据，而是索引一定范围的数据。也就是从已排序的全量数据中，间隔性的选取一些数据记录主键字段的值来生成primary.idx索引文件，从而加快表查询效率。间隔设置参数为index_granularity。

我们向表test中插入9条数据，

insert into test.test(id, type, create_time) VALUES (1, 1, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (1, 2, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (1, 3, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (2, 1, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (2, 1, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (3, 1, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (3, 1, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (4, 1, toDateTime('2021-03-01 00:00:00')); 
insert into test.test(id, type, create_time) VALUES (5, 1, toDateTime('2021-03-01 00:00:00'));

因为index_granularity设置为4，所以每4条数据就会生成一条索引记录，即使用插入的第1、5、9条数据id字段的值生成索引文件记录。

2.3.3 标记文件

mrk标记文件在primary.idx索引文件和bin数据文件之间起到了桥梁作用。primary.idx文件中的每条索引在mrk文件中都有对应的一条记录。一条记录的组成包括：

offset-compressed bin file：表示指向的压缩数据块在bin文件中的偏移量。
offset-decompressed data block：表示指向的数据在解压数据块中的偏移量。
row counts：代表数据记录行数，小于等于index_granularity所设置的值。

索引，标记和数据文件下图所示：

clickhouse通过jdbc实现批量数据导入代码示例

Tue, 09 Jan 2024 14:41:53 +0800

通过mybatis也可以实现批量写入clickhouse，但是效率太低，每秒大概只能写入300条数据，对于动辄上千万的数据或者更多数据，效率就太低了。而采用jdbc驱动的方式进行批量写入，每秒大约写入数据量达到10000条，效率得到大幅度的提升，只是这种方式，需要编写的代码会更多，需要处理的细节也更多。

1. 添加依赖

<!--clickhouse-->
<dependency>
<groupId>com.clickhouse</groupId>
<artifactId>clickhouse-jdbc</artifactId>
<version>0.3.2-patch8</version>
</dependency>

2. 获取连接

public Connection getClickHouseConnection(){
        
	String url = "jdbc:clickhouse://192.168.17.81:8123/default";
 
	Properties props = new Properties();
	props.setProperty("user", username);
	props.setProperty("password", "123456");	
 
	try {
		
		ClickHouseDataSource dataSource = new ClickHouseDataSource(url, props);
		Connection conn = dataSource.getConnection();
		return conn;
	}catch (Exception e)
	{
		e.printStackTrace();
	}
 
	return null;	
}
 
 
 
// 关闭连接
public void closeClickHouseConnection(Connection connection)
{
	if (connection != null) {
		try {
			connection.close();
		} catch (SQLException e) {
			//ignore
		}
	}
}

3. 准备预编译sql

private String buildBatchSql()
{
	String baseSql = "INSERT INTO plat_data_center_bill(bill_id, bank_merchno, bank_sub_merchno, ...) ";
	baseSql = baseSql + "VALUES(";
 
	baseSql = baseSql + "?,?,?,?,?,?,?,?,?,?";
	baseSql = baseSql + ",?,?,?,?,?,?,?,?,?,?";        
 
	baseSql = baseSql + ")";
 
	return baseSql;
}

注意这里的字段对应的值采用问号替换，一个字段就是一个问号。

4. 批量写入数据

private void batchInsertClickHouse(Connection connection, List<PlatDataCenterBill> dataList)
{        
	String batchSql = buildBatchSql();
 
	PreparedStatement prepareStatement = null;
	try {
		 connection.setAutoCommit(true);
		// 执行conn.setAutoCommit(false);会报Transactions are not supported异常
		// 所以不能执行conn.commit();
		// 只能执行pst.executeBatch();由clickhouse进行后台提交。测试时插入数据条数是正确的
		prepareStatement = connection.prepareStatement(batchSql);
		for (PlatDataCenterBill bill : dataList) {
			prepareStatement.setLong(1, bill.getBillId());
 
			setPrepareString(prepareStatement, 2, bill.getBankMerchno());
 
			setPrepareString(prepareStatement,2, bill.getBankMerchno());
			setPrepareString(prepareStatement,3, bill.getBankSubMerchno());
			setPrepareString(prepareStatement,4, bill.getBankMerchname());
			prepareStatement.setObject(5, bill.getBankDeptid());
 
			setPrepareString(prepareStatement,6, bill.getBankDeptances());
			setPrepareString(prepareStatement,7, bill.getBankDeptname());
			setPrepareString(prepareStatement,8, bill.getAppId());
			setPrepareString(prepareStatement,9, bill.getExternalMerchno());
			setPrepareString(prepareStatement,10, bill.getExternalMerchname());
 
			...		
 
			prepareStatement.addBatch();
		}
		prepareStatement.executeBatch();
		prepareStatement.clearBatch();
		
	} catch (Exception e) {
		e.printStackTrace();
		throw new ServiceException("批量写入异常");
	} finally {
		if (prepareStatement != null) {
			try {
				prepareStatement.close();
			} catch (SQLException e) {
				//ignore
			}
		}    
 
	}
}

这里需要注意的是需要对每个字段的值进行设置，即使字段值为空值，也需要设置，另外还需要注意，日期时间类型，需要使用Timestamp进行设置。

prepareStatement.setObject(48, new Timestamp(sqlCreateTime.getTime()));

5. 为了判断空值，使用统一的方法

public void setPrepareString(PreparedStatement pst, int idx, String value) throws SQLException {
 
	if(value != null) {
		pst.setString(idx, value);
	}
	else  {
		pst.setString(idx, "");
	}
}

6. 相关的依赖

import com.clickhouse.jdbc.ClickHouseDataSource;
 
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
 
import javax.sql.DataSource;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

7. 循环批量写数据

Connection connection = getClickHouseConnection();
 
 
List<PlatDataCenterBill> platBills = pdcBillService.selectPlatDataCenterBillList(pdcbill);
 
int iCount = 0;
 
ArrayList<PlatDataCenterBill> billList = new ArrayList<PlatDataCenterBill>();
 
for(PlatDataCenterBill bill : platBills)
{
	billList.add(bill);
	iCount = iCount + 1;
 
	if(iCount % 10000 == 0)
	{		
		System.out.println(billList.size());
		batchInsertClickHouse(connection, billList);
 
		billList.clear();
	}
}
 
// 剩余部分数据批量写入
if(billList.size() > 0)
{	
	batchInsertClickHouse(connection, billList);
	billList.clear();
}
 
 
// 这里关闭连接才是正确的
closeClickHouseConnection(connection);

以上就是批量导入数据的大致处理过程。

Clickhouse MergeTree引擎配置说明

Wed, 27 Dec 2023 22:13:02 +0800

总结

分区目录可以不断被合并，最终可能被合并成一个目录
分区目录数据量最大可达150G
插入的数据只有落地到磁盘之后，客户端才能查询到数据
数据写入类似于kafka：支持同步，异步写入。支持最近数据插入去重校验
数据块合并之后，旧的数据块默认情况下会在480s之后被删除

注意：

不活动的数据块：数据被合并后旧的数据块
活动的数据块：合并新生成或刚写入的数据块
数据块就是：表数据目录下生成的存储数据的目录

配置说明

max_parts_in_total

如果表的所有分区中的活动数据块总数超过 max_parts_in_total，则插入操作将被中断并抛出“Too many parts (N)”异常。

可能的取值：

任何正整数。
默认值为 100000。

表中大量的数据块会降低 ClickHouse 查询的性能并增加 ClickHouse 的启动时间。

parts_to_throw_insert

如果单个分区中活动数据块数量超过parts_to_throw_insert值，则插入操作会因为Too many parts（N）而中断。与插入操作相比，

合并操作的处理速度显著较慢。

可能的取值：

任何正整数。
默认值：300。

为了实现SELECT查询的最大性能，需要尽量减少分区数据块的数量，请参见Merge Tree。

您可以将值设置为更大，这将降低Too many parts错误的概率，但同时SELECT性能可能会下降。

parts_to_delay_insert

如果单个分区中的活动数据块数量超过 parts_to_delay_insert，则插入操作会人为地减慢。

可能的取值：

任何正整数。
默认值为 150。

ClickHouse 人为地延长了 INSERT 的执行时间（添加“sleep”），提升后天数据合并效率。

max_delay_to_insert

如果单个分区中的活动数据块数量超过 parts_to_throw_insert，则此参数用于计算插入延迟的秒数值。

可能的取值：

任何正整数。
默认值为 1。

INSERT 的延迟（以毫秒为单位）通过以下公式计算：

max_k = parts_to_throw_insert - parts_to_delay_insert

k = 1 + parts_count_in_partition - parts_to_delay_insert

delay_milliseconds = pow(max_delay_to_insert * 1000, k / max_k)

例如，如果一个分区有 299 个活动数据块，parts_to_throw_insert 为 300，parts_to_delay_insert 为 150，max_delay_to_insert 为 1，

则插入将延迟 pow(1 * 1000, (1 + 299 - 150) / (300 - 150)) = 1000 毫秒。

从版本 23.1 开始，公式已更改为：

allowed_parts_over_threshold = parts_to_throw_insert - parts_to_delay_insert

parts_over_threshold = parts_count_in_partition - parts_to_delay_insert + 1

delay_milliseconds = max(min_delay_to_insert_ms, (max_delay_to_insert * 1000) * parts_over_threshold / allowed_parts_over_threshold)

例如，如果一个分区有 224 个活动数据块，parts_to_throw_insert 为 300，parts_to_delay_insert 为 150，max_delay_to_insert 为 1，

min_delay_to_insert_ms 为 10，则插入将延迟 max(10, 1 * 1000 * (224 - 150 + 1) / (300 - 150)) = 500 毫秒

inactive_parts_to_throw_insert

如果单个分区中的非活动数据块数量超过 inactive_parts_to_throw_insert，则插入操作将被中断并抛出“Too many inactive parts (N).

Parts cleaning are processing significantly slower than inserts”异常。

可能的取值：

任何正整数。
默认值为 0（不限制）。

inactive_parts_to_delay_insert

如果表中单个分区中的非活动数据块数量至少达到 inactive_parts_to_delay_insert，则插入操作会人为地减慢。当服务器无法快速清理数据块时，

这很有用。

可能的取值：

任何正整数。
默认值为 0（不限制）。

simultaneous_parts_removal_limit

如果存在大量过时的数据块，则清理线程将尝试在一次迭代中删除多达 simultaneous_parts_removal_limit 个数据块。

将 simultaneous_parts_removal_limit 设置为 0 表示没有限制。

默认值为 0。

replicated_deduplication_window

写入副本表时存储数据块哈希值以防止重复数据插入。由于网络抖动，客户端可能会发起两次数据插入，每次数据插入节点可能不一样，

所以 ClickHouse 将数据块的哈希值写入 ClickHouse Keeper，用于副本节点之间共享数据块的哈希值

可能的取值：

任何正整数。
0（禁用去重）。
默认值为 100。

Insert 命令会创建一个或多个数据块（part）。写入副本表时会进行插入去重校验，ClickHouse 将所创建数据块的哈希值写入 ClickHouse Keeper。

仅存储最近的 replicated_deduplication_window 个数据块的哈希值。旧的哈希值将从 ClickHouse Keeper 中删除。

如果replicated_deduplication_window 设置过大，会减慢插入，因为需要比较更多的条目。

哈希值是从插入数据的字段名称、类型和数据（字节流）通过hash计算而来的。

non_replicated_deduplication_window

在非副本表的 MergeTree 表中，存储用于检查重复插入数据块的哈希值的个数

可能的取值：

任何正整数。
0（禁用去重）。
默认值为 0。

使用与副本表类似的去重机制（参见 replicated_deduplication_window 设置）

replicated_deduplication_window_seconds

数据块的哈希值从 ClickHouse Keeper 中删除的时间（以秒为单位）。

可能的取值：

任何正整数。
默认值为 604800（1 周）。

与 replicated_deduplication_window 类似，replicated_deduplication_window_seconds 指定存储插入去重的数据块的哈希值保留时长。

超过 replicated_deduplication_window_seconds 的哈希值将从 ClickHouse Keeper 中删除，即使数量小于 replicated_deduplication_window。

如果只有一条记录，则会永久存储。

replicated_deduplication_window_for_async_inserts

类似配置：replicated_deduplication_window，该配置应用于异步数据插入并且默认值为 10000。

replicated_deduplication_window_seconds_for_async_inserts

类似配置：replicated_deduplication_window_seconds

use_async_block_ids_cache

如果为 true，则缓存异步插入数据块的哈希值，提升插入性能。

可能的取值：

true，false
默认值为 false。

一个包含多个异步插入将生成多个数据块的哈希值。这些哈希值存储在Keeper 上，如果开启这个缓存，

Clickhouse将监视 Keeper 中的存储该分区哈希值的路径。如果在 Keeper 中监视到更新，则尽快更新缓存，以便我们能够在内存中过滤重复的插入。

async_block_ids_cache_min_update_interval_ms

更新数据块哈希值的最小间隔（以毫秒为单位）。

可能的取值：

任何正整数。
默认值为 100。

通常，只要监视 Keeper 路径中有更新，内存中的数据块哈希值就会立即更新。但是，缓存更新可能过于频繁并成为重负担。

这个最小间隔可以防止缓存更新得太快。

max_replicated_logs_to_keep

How many records may be in the ClickHouse Keeper log if there is inactive replica.

An inactive replica becomes lost when when this number exceed.

Possible values:

Any positive integer.
Default value: 1000

min_replicated_logs_to_keep

同上，这是最小值，默认值为：10

prefer_fetch_merged_part_time_threshold

在副本同步过程中，如果zk上同步日志创建时长大于prefer_fetch_merged_part_time_threshold，并且需要同步的数据大小总和大于

prefer_fetch_merged_part_size_threshold，则优先从副本中拉取合并后的数据块，而不是拉取原始数据并在本地执行合并。这是为了加速数据合并。

可能的取值：

任何正整数。
默认值为 3600。

prefer_fetch_merged_part_size_threshold

在副本同步过程中，如果zk上同步日志创建时长大于prefer_fetch_merged_part_time_threshold，并且需要同步的数据大小总和大于

prefer_fetch_merged_part_size_threshold，则优先从副本中拉取合并后的数据块，而不是拉取原始数据并在本地执行合并。这是为了加速数据合并。

可能的取值：

任何正整数。
默认值为 10,737,418,240（10G）

execute_merges_on_single_replica_time_threshold

当此设置的值大于零时，则仅一个副本立即开始合并，其他副本等待execute_merges_on_single_replica_time_threshold秒后下载合并好的数据。

如果所选副本在execute_merges_on_single_replica_time_threshold秒后未完成合并，则会回退到标准行为。

可能的取值：

任何正整数。
默认值为0（秒）。

remote_fs_execute_merges_on_single_replica_time_threshold

当此设置的值大于零时，如果合并数据块在共享存储上（比如s3上），并且启用了 allow_remote_fs_zero_copy_replication，

则仅一个副本立即开始合并。其他的副本等待，超过该值还没有完成合并，则会回退到标准行为。

allow_remote_fs_zero_copy_replication 还处于实验阶段

ClickHouse 版本 22.8 及更高版本默认禁用。不建议在生产环境中使用此功能。

可能的取值：

任何正整数。
默认值为 10800。

try_fetch_recompressed_part_timeout

Recompression 是代表更换压缩方式对数据进行压缩？？

Recompression works slow in most cases, so we don’t start merge with recompression until this timeout and trying to

fetch recompressed part from replica which assigned this merge with recompression.

Possible values:

Any positive integer.
Default value: 7200

always_fetch_merged_part

副本同步时，只拉取合并好的分区

可能的取值：

true，false
默认值为 false。

max_suspicious_broken_parts

分区中最大的被损坏的数据块数量，如果超过此值，则停止自动删除。

可能的取值：

任何正整数。
默认值为 100。

max_suspicious_broken_parts_bytes

分区中最大被损坏的数据块的数据总和大小，如果超过此值，则停止自动删除。

可能的取值：

任何正整数。
默认值为 1,073,741,824。（1G）

max_files_to_modify_in_alter_columns

如果要修改（删除、添加）的文件数量大于此设置，则取消ALTER操作。

可能的取值：

任何正整数。
默认值为75。

max_files_to_remove_in_alter_columns

如果在列删除时，需要删除的文件数量大于此设置，则取消ALTER操作。

可能的取值：

任何正整数。
默认值为50。

replicated_max_ratio_of_wrong_parts

如果被损坏的数据块占总数据块的数量的比例小于此值，则允许启动，反之就无法启动

可能的取值：

浮点数，0.0 - 1.0。
默认值为0.5。

replicated_max_parallel_fetches_for_host

副本数据同步线程次大小

可能的取值：

任何正整数。
默认值为15。
replicated_can_become_leader
标记节点是否参与主副本的竞争

可能的取值：

true，false
默认值为 true。

zookeeper_session_expiration_check_period

ZooKeeper 会话过期检查周期，以秒为单位。

可能的取值：

任何正整数。
默认值为60。

detach_old_local_parts_when_cloning_replica

在修复丢失的副本时，不要删除旧的本地数据块。

可能的取值：

true，false
默认值为 true。

replicated_fetches_http_connection_timeout

拉取数据数据块请求的连接超时（以秒为单位）。如果未明确设置，则继承自默认配置文件的 http_connection_timeout。

可能的取值：

任何正整数。
0 - 使用 http_connection_timeout 的值。
默认值为0。

replicated_fetches_http_send_timeout

拉取数据块的请求发送超时时间（以秒为单位。如果未明确设置，则继承自默认配置文件的 http_send_timeout。

可能的取值：

任何正整数。
0 - 使用 http_send_timeout 的值。
默认值为0。

replicated_fetches_http_receive_timeout

拉取数据块的请求接收超时时间（以秒为单位）。如果未明确设置，则继承自默认配置文件的 http_receive_timeout。

可能的取值：

任何正整数。
0 - 使用 http_receive_timeout 的值。
默认值为0

max_replicated_fetches_network_bandwidth

限制副本同步过程中数据拉取最大速度，以每秒字节数为单位。这个设置可以应用于特定的表，

max_replicated_fetches_network_bandwidth_for_server用于控制Clickhouse server限制副本同步过程中数据拉取最大速度，但是对于这个设置，

表级别的值应该小于服务器级别的值。否则，服务器只考虑 max_replicated_fetches_network_bandwidth_for_server

这个设置不能完全精确地跟踪。

可能的取值：

正整数。
0 - 无限制。
默认值为0。

用法

可以用于在向新增或替换节点，复制数据时限制速度。

max_replicated_sends_network_bandwidth

限制副本同步过程中数据发送出去最大速度，以每秒字节数为单位。这个设置可以应用于特定的表，

max_replicated_sends_network_bandwidth_for_server 用于控制Clickhouse server限制副本同步过程中数据发送出去最大速度，但是对于这个设置，表级别的值应该小于服务器级别的值。否则，服务器只考虑 max_replicated_sends_network_bandwidth_for_server

这个设置不能完全精确地跟踪。

可能的取值：

正整数。
0 - 无限制。
默认值为0。

用法

可以用于在向新增或替换节点，复制数据时限制速度。

old_parts_lifetime

合并后产生的非活动数据块保留时长（以秒为单位）。

可能的取值：

任何正整数。
默认值为：480。

多个数据块合并为一个新数据块后，ClickHouse 将原始数据块标记为非活动状态，如果它们不被现有的查询引用，即数据块的引用计数为零，

在 old_parts_lifetime 秒后会删除它们。

对于新的数据块，不会调用 fsync，因此新的数据块在一段时间内仅存在于服务器的内存（操作系统缓存）中。如果服务器异常重启，

新的数据块可能会丢失或损坏。为了保护数据，非活动数据块不会立即被删除。

在启动期间，ClickHouse 会检查数据块的完整性。如果合并的数据块已损坏，则将原来非活动数据块放到活动列表中，并在以后再次合并它们。

然后，将已损坏的数据块重命名（添加 broken_ 前缀）并移动到 detached 文件夹中。如果合并的数据块未损坏，

则将原始非活动数据块重命名（添加 ignored_ 前缀）并移动到 detached 文件夹中。

默认的 dirty_expire_centisecs 值（Linux 内核设置）为 30 秒（需要被写入磁盘数据存储在内存中的最长时间），

但在磁盘系统的负载高的情况下，数据可以写入得更晚。通过实验，选择了 old_parts_lifetime 的值为 480 秒，这样可以保证新数据块被写入磁盘

max_bytes_to_merge_at_max_space_in_pool

如果有足够的资源可用的情况下，合并为一个数据块的最大总数据块大小（以字节为单位），max_bytes_to_merge_at_max_space_in_pool

大致对应于自动后台合并后最大可能数据块大小。

可能的取值：

任何正整数。
默认值为161061273600（150 GB）。

合并调度程序会定期分析分区中的数据块大小和数量，如果池中有足够的空闲资源，则启动后台合并。合并将持续进行，

直到源数据块的总大小大于 max_bytes_to_merge_at_max_space_in_pool。

由 OPTIMIZE FINAL 发起的合并会忽略 max_bytes_to_merge_at_max_space_in_pool，只考虑可用资源（空闲磁盘空间）进行合并，

直到分区中只剩下一个数据块为止。

max_bytes_to_merge_at_min_space_in_pool

可用资源最少的情况，合并为一个数据块的最小总数据块大小（以字节为单位）

可能的取值：

任何正整数。
默认值为1048576（1 MB）。

max_bytes_to_merge_at_min_space_in_pool 定义了即使缺少可用磁盘空间，也可以合并的数据块的最大总大小。

这是为了减少小数据块的数量和“Too many parts”错误的几率。由于合并过程中会翻倍占用磁盘空间。

因此，当可用磁盘空间很少时，可能会出现这样的情况：虽然有空闲空间，但该空间已经被正在进行的大型合并所占用，

因此其他合并无法启动，并且随着大量的数据插入，小数据块的数量会不断增多。

merge_max_block_size

合并过程中每次从原始数据中读取的行数。

可能的取值：

任何正整数。
默认值为8192。

数据合并时，将从原始数据中以 merge_max_block_size 行作为一个段进行读取，然后将数据合并写入到新的数据块。

读取的数据会被放置在内存中，因此 merge_max_block_size 影响合并所需的内存大小。

因此，对于具有多列的表（如果平均行大小为100kb，则合并10个数据块时，（100kb * 10 * 8192）= ~8GB RAM），合并可能会消耗大量内存。

通过减小 merge_max_block_size，可以减少合并所需的内存，但会减低合并操作性能。

number_of_free_entries_in_pool_to_lower_max_size_of_merge

当池中（或复制队列中）的可用空闲条目少于number_of_free_entries_in_pool_to_lower_max_size_of_merge时，开始减少每次合并的数据量，

减少合并线程占用太多时间

可能的取值：

任何正整数。
默认值为8。

number_of_free_entries_in_pool_to_execute_mutation

当线程池中空闲数少于number_of_free_entries_in_pool_to_execute_mutation时，不再执行数据变更操作。

这是为了留出空闲线程进行常规合并操作，避免出现“Too many parts”的错误。

可能的取值：

任何正整数。
默认值为20。

使用：

number_of_free_entries_in_pool_to_execute_mutation

设置的值应该小于 background_pool_size * background_merges_mutations_concurrency_ratio 的值。否则，ClickHouse 会抛出异常。

min_age_to_force_merge_seconds

Merge parts if every part in the range is older than the value of min_age_to_force_merge_seconds.

Possible values:

Positive integer.
Default value: 0 — Disabled.

min_age_to_force_merge_on_partition_only

参数 min_age_to_force_merge_seconds是否仅应用于整个分区，而不是应用子目录上（project 目录）。

可能的取值：

true，false
默认值为 false。

max_part_loading_threads

加载分区元数据的线程数，默认与机器cup数量相等，在启动期间，ClickHouse 会读取所有表数据块的元数据文件，以构建内存中数据块的列表。

如果数据块比较多，这个过程可能需要很长时间，可以通过适当增加max_part_loading_threads来缩短这个启动时间。

可能的取值：

任何正整数。
默认值为 auto（即 CPU 核心数）。

max_partitions_to_read

限制一次查询中可以访问的最大分区数。可以通过查询级别的设置来覆盖表创建时指定的设置值。

可能的取值：

任何正整数。
默认值为-1（无限制）

allow_floating_point_partition_key

允许浮点数类型字段作为分区字段

可能值:

0 - 不允许使用浮点分区键。
1 - 允许使用浮点分区键。
默认值: 0。

check_sample_column_is_correct

Enables the check at table creation, that the data type of a column for sampling or sampling expression is correct.

The data type must be one of unsigned integer types: UInt8, UInt16, UInt32, UInt64.

Possible values:

true — The check is enabled.
false — The check is disabled at table creation.
Default value: true.

By default, the ClickHouse server checks at table creation the data type of a column for sampling or sampling expression.

If you already have tables with incorrect sampling expression and do not want the server to raise an exception during startup,

set check_sample_column_is_correct to false.

min_bytes_to_rebalance_partition_over_jbod

每次写入某个磁盘阵列时，数据的最小值

可能取值为：

正整数。
0 - 禁用平衡。
默认值为0。

使用：

min_bytes_to_rebalance_partition_over_jbod设置的值应该大于max_bytes_to_merge_at_max_space_in_pool/1024的值。

否则，ClickHouse会抛出异常。

detach_not_byte_identical_parts

副本之间数据同步完成后，发现数据与副本数据不一样，是把副本数据卸载掉，还是直接删掉这些数据

可能的取值：

0 - 删除数据块。
1 - 卸载数据块。
默认值为0。

merge_tree_clear_old_temporary_directories_interval_seconds

设置多长时间（单位：秒）清理无用临时目录

可能的取值：

任何正整数。
默认值为60秒。

merge_tree_clear_old_parts_interval_seconds

设置多长时间（单位：秒）清理无用数据

可能的取值：

任何正整数。
默认值为1秒。

max_concurrent_queries

最大查询并发量，收到其他级别max_concurrent_queries的限制，比如整个集群查询最大并发量是100，表的最大查询并发量不能超过100

可能的取值：

正整数。
0 - 没有限制。
默认值为0（没有限制）。

min_marks_to_honor_max_concurrent_queries

The minimal number of marks read by the query for applying the max_concurrent_queries setting.

Note that queries will still be limited by other max_concurrent_queries settings.

Possible values:

Positive integer.
0 — Disabled (max_concurrent_queries limit applied to no queries).
Default value: 0 (limit never applied).

ratio_of_defaults_for_sparse_serialization

列中默认值数量与所有值数量之比的最小值。设置该值主要是为了使用稀疏序列化存储，提升查询效率。

如果列是稀疏的（比如大部分行都是0），ClickHouse可以使用稀疏格式对其进行编码，在查询过程中数据不需要进行完全解压缩提升查询效率。

要启用此稀疏序列化压缩特性，请将ratio_of_defaults_for_sparse_serialization设置为小于1.0的值。

如果该值大于或等于1.0（默认值），则使用常规的方法序列化列数据。

可能的取值：

介于0和1之间的小数，表示启用稀疏序列化特性。
如果不想使用稀疏序列化，则设置为1.0或更大。
默认值为1.0（表示禁用稀疏序列化特性）。

案例

以下表中的s列有95%的行为空字符串。在my_regular_table中，我们不使用稀疏序列化，在my_sparse_table中，我们将ratio_of_defaults_for_sparse_serialization设置为0.95

CREATE TABLE my_regular_table
(
    `id` UInt64,
    `s` String
)
ENGINE = MergeTree
ORDER BY id;

INSERT INTO my_regular_table
SELECT
    number AS id,
    number % 20 = 0 ? toString(number): '' AS s
FROM
    numbers(10000000);


CREATE TABLE my_sparse_table
(
    `id` UInt64,
    `s` String
)
ENGINE = MergeTree
ORDER BY id
SETTINGS ratio_of_defaults_for_sparse_serialization = 0.95;

INSERT INTO my_sparse_table
SELECT
    number,
    number % 20 = 0 ? toString(number): ''
FROM
    numbers(10000000);

您可以通过查看system.parts_columns表的serialization_kind列来验证列是否使用稀疏编码：

SELECT column, serialization_kind FROM system.parts_columns
WHERE table LIKE 'my_sparse_table';
┌─column─┬─serialization_kind─┐
│ id     │ Default            │
│ s      │ Default            │
│ id     │ Default            │
│ s      │ Sparse             │
│ id     │ Default            │
│ s      │ Sparse             │
└────────┴────────────────────┘

clean_deleted_rows

启用/禁用使用ReplacingMergeTree引擎的表上执行OPTIMIZE … FINAL时自动删除标记为is_deleted的行。当禁用时，

必须在OPTIMIZE … FINAL中添加CLEANUP关键字才能获得相同的行为。

可选的值

Always 或者 Never
默认：Never

参考资料

https://clickhouse.com/docs/en/operations/settings/settings

Springboot集成ClickHouse

Fri, 22 Dec 2023 09:46:24 +0800

1、clickhouse应⽤场景

1.绝大多数请求都是用于读访问的

2.数据需要以大批次（大于1000行）进行更新，而不是单行更新；或者根本没有更新操作

3.数据只是添加到数据库，没有必要修改

4.读取数据时，会从数据库中提取出大量的行，但只用到一小部分列

5.表很“宽”，即表中包含大量的列

6.查询频率相对较低（通常每台服务器每秒查询数百次或更少）

7.对于简单查询，允许大约50毫秒的延迟

8.列的值是比较小的数值和短字符串（例如，每个URL只有60个字节）

9.在处理单个查询时需要高吞吐量（每台服务器每秒高达数十亿行）

10.不需要事务

11.数据一致性要求较低

12.每次查询中只会查询一个大表。除了一个大表，其余都是小表

13.查询结果显著小于数据源。即数据有过滤或聚合。返回结果不超过单个服务器内存大小

分析类查询，通常只需要读取表的一小部分列。在列式数据库中可以只读取需要的数据。数据总是打包成批量读取的，所以压缩是非常容易的。同时数据按列分别存储这也更容易压缩。这进一步降低了I/O的体积。由于I/O的降低，这将帮助更多的数据被系统缓存。

2、整合Springboot：

核心依赖（mybatis plus做持久层，druid做数据源）：

<dependencies>
    <!--clickhouse-->
    <dependency>
        <groupId>ru.yandex.clickhouse</groupId>
        <artifactId>clickhouse-jdbc</artifactId>
        <version>0.3.1-patch</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/com.alibaba/druid -->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid</artifactId>
        <version>1.2.6</version>
    </dependency>
    <!-- https://mvnrepository.com/artifact/com.baomidou/mybatis-plus-boot-starter -->
    <dependency>
        <groupId>com.baomidou</groupId>
        <artifactId>mybatis-plus-boot-starter</artifactId>
        <version>3.4.3.1</version>
    </dependency>
</dependencies>

配置yml文件：

spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    click:
      driverClassName: ru.yandex.clickhouse.ClickHouseDriver
      url: jdbc:clickhouse://127.0.0.1:8123/dbname
      username: username
      password: 123456
      initialSize: 10
      maxActive: 100
      minIdle: 10
      maxWait: 6000
 
mybatis-plus:
  mapper-locations: classpath*:mapper/*.xml
  configuration:
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
    map-underscore-to-camel-case: true
    cache-enabled: true
    lazy-loading-enabled: true
    multiple-result-sets-enabled: true
    use-generated-keys: true
    default-statement-timeout: 60
    default-fetch-size: 100
  type-aliases-package: com.example.tonghp.entity

ClickHouse与Druid连接池配置类：

参数配置：

package com.example.tonghp.config;
 
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
 
/**
 * @author: tonghp
 * @create: 2021/07/26 16:23
 */
@Data
@Component
@ConfigurationProperties(prefix = "spring.datasource.click")
public class JdbcParamConfig {
    private String driverClassName ;
    private String url ;
    private Integer initialSize ;
    private Integer maxActive ;
    private Integer minIdle ;
    private Integer maxWait ;
    private String username;
    private String password;
    // 省略 GET 和 SET
}

Druid连接池配置

package com.example.tonghp.config;
 
import com.alibaba.druid.pool.DruidDataSource;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
import javax.annotation.Resource;
import javax.sql.DataSource;
import javax.swing.*;
 
/**
 * @author: tonghp
 * @create: 2021/07/26 16:22
 */
@Configuration
public class DruidConfig {
 
    @Resource
    private JdbcParamConfig jdbcParamConfig ;
 
    @Bean
    public DataSource dataSource() {
        DruidDataSource datasource = new DruidDataSource();
        datasource.setUrl(jdbcParamConfig.getUrl());
        datasource.setDriverClassName(jdbcParamConfig.getDriverClassName());
        datasource.setInitialSize(jdbcParamConfig.getInitialSize());
        datasource.setMinIdle(jdbcParamConfig.getMinIdle());
        datasource.setMaxActive(jdbcParamConfig.getMaxActive());
        datasource.setMaxWait(jdbcParamConfig.getMaxWait());
        datasource.setUsername(jdbcParamConfig.getUsername());
        datasource.setPassword(jdbcParamConfig.getPassword());
        return datasource;
    }
}

接下来配置实体类，mapper，service，controlle以及mapper.xml。与mybatisplus操作mysql

package com.example.tonghp.entity;
 
import lombok.Data;
 
import java.io.Serializable;
 
/**
 * @author: tonghp
 * @create: 2021/07/26 16:31
 */
@Data
public class UserInfo implements Serializable {
    private static final long serialVersionUID = 1L;
    private int id;
    private String userName;
    private String passWord;
    private String phone;
    private String email;
    private String createDay;
}

package com.example.tonghp.mapper;
 
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.example.tonghp.entity.UserInfo;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Repository;
 
import java.util.List;
 
/**
 * @author: tonghp
 * @create: 2021/07/26 16:32
 */
@Repository
public interface UserInfoMapper extends BaseMapper<UserInfo> {
    // 写入数据
    void saveData (UserInfo userInfo) ;
    // ID 查询
    UserInfo selectById (@Param("id") Integer id) ;
    // 查询全部
    List<UserInfo> selectList () ;
}

UserInfoMapper.xml

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.example.tonghp.mapper.UserInfoMapper">
 
    <resultMap id="BaseResultMap" type="com.example.tonghp.entity.UserInfo">
        <id column="id" jdbcType="INTEGER" property="id" />
        <result column="user_name" jdbcType="VARCHAR" property="userName" />
        <result column="pass_word" jdbcType="VARCHAR" property="passWord" />
        <result column="phone" jdbcType="VARCHAR" property="phone" />
        <result column="email" jdbcType="VARCHAR" property="email" />
        <result column="create_day" jdbcType="VARCHAR" property="createDay" />
    </resultMap>
    <sql id="Base_Column_List">
        id,user_name,pass_word,phone,email,create_day
    </sql>
    <insert id="saveData" parameterType="com.example.tonghp.entity.UserInfo" >
        INSERT INTO cs_user_info
        (id,user_name,pass_word,phone,email,create_day)
        VALUES
        (#{id,jdbcType=INTEGER},#{userName,jdbcType=VARCHAR},#{passWord,jdbcType=VARCHAR},
        #{phone,jdbcType=VARCHAR},#{email,jdbcType=VARCHAR},#{createDay,jdbcType=VARCHAR})
    </insert>
    <select id="selectById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
        select
        <include refid="Base_Column_List" />
        from cs_user_info
        where id = #{id,jdbcType=INTEGER}
    </select>
    <select id="selectList" resultMap="BaseResultMap" >
        select
        <include refid="Base_Column_List" />
        from cs_user_info
    </select>
 
</mapper>

Service

package com.example.tonghp.service;
 
import com.baomidou.mybatisplus.extension.service.IService;
import com.example.tonghp.entity.UserInfo;
import org.apache.ibatis.annotations.Param;
import org.springframework.stereotype.Service;
 
import java.util.List;
 
/**
 * @author: tonghp
 * @create: 2021/07/26 16:46
 */
public interface UserInfoService extends IService<UserInfo> {
    // 写入数据
    void saveData (UserInfo userInfo) ;
    // ID 查询
    UserInfo selectById (@Param("id") Integer id) ;
    // 查询全部
    List<UserInfo> selectList () ;
}

ServiceImpl

package com.example.tonghp.service.impl;
 
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.example.tonghp.entity.UserInfo;
import com.example.tonghp.mapper.UserInfoMapper;
import com.example.tonghp.service.UserInfoService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
import java.util.List;
 
/**
 * @author: tonghp
 * @create: 2021/07/26 16:48
 */
 
@Service
public class UserInfoServiceImpl extends ServiceImpl<UserInfoMapper, UserInfo> implements UserInfoService {
 
    @Autowired
    UserInfoMapper userInfoMapper;
 
    @Override
    public void saveData(UserInfo userInfo) {
        userInfoMapper.saveData(userInfo);
    }
 
    @Override
    public UserInfo selectById(Integer id) {
        return userInfoMapper.selectById(id);
    }
 
    @Override
    public List<UserInfo> selectList() {
        return userInfoMapper.selectList();
    }
}

Controller

package com.example.tonghp.controller;
 
import com.example.tonghp.entity.UserInfo;
import com.example.tonghp.service.UserInfoService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
 
import javax.annotation.Resource;
import java.util.List;
 
/**
 * @author: tonghp
 * @create: 2021/07/26 16:45
 */
@RestController
@RequestMapping("user")
public class UserInfoController {
 
    @Autowired
    private UserInfoService userInfoService ;
 
    @RequestMapping("saveData")
    public String saveData (){
        UserInfo userInfo = new UserInfo () ;
        userInfo.setId(4);
        userInfo.setUserName("winter");
        userInfo.setPassWord("567");
        userInfo.setPhone("13977776789");
        userInfo.setEmail("winter");
        userInfo.setCreateDay("2020-02-20");
        userInfoService.saveData(userInfo);
        return "sus";
    }
 
    @RequestMapping("selectById")
    public UserInfo selectById () {
        return userInfoService.selectById(1) ;
    }
 
    @RequestMapping("selectList")
    public List<UserInfo> selectList () {
        return userInfoService.selectList() ;
    }
}

main()

package com.example.tonghp;
 
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
 
@SpringBootApplication
@MapperScan("com.example.tonghp.mapper")
public class TonghpApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(TonghpApplication.class, args);
    }
}

使用jdbc方式操作clickhouse

1、创建表及数据

create table t_order01(
 id UInt32,
 sku_id String,
 total_amount Decimal(16,2),
 create_time Datetime
) engine =MergeTree
 partition by toYYYYMMDD(create_time)
 primary key (id)
 order by (id,sku_id);
   
insert into t_order01 values
(101,'sku_001',1000.00,'2021-12-01 12:00:00'),
(102,'sku_002',2000.00,'2021-12-01 11:00:00'),
(102,'sku_004',2500.00,'2021-12-01 12:00:00'),
(102,'sku_002',2000.00,'2021-12-01 13:00:00'),
(102,'sku_002',12000.00,'2021-12-01 13:00:00'),
(102,'sku_002',600.00,'2020-06-12 12:00:00');

2、引入clickhouse的jdbc依赖

<dependency>
    <groupId>ru.yandex.clickhouse</groupId>
    <artifactId>clickhouse-jdbc</artifactId>
    <version>0.1.52</version>
</dependency>

3、实例代码

实大部分的操作和我们使用jdbc操作mysql的步骤类似，下面直接贴出代码，可以结合注释进行参考使用

import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.settings.ClickHouseProperties;
   
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
   
public class CreateTableTest {
   
    private static String username = "default";
    private static String password = "你的连接密码";
    private static String address = "jdbc:clickhouse://clickhouse的连接IP地址:8123";
    private static String db = "连接数据库名称（默认数据库：default）";
    private static int socketTimeout = 600000;
   
    public static void main(String[] args) throws Exception {
        //getConn();
        //queryTable();
        //createTable("");
        //insertOne();
        //dropTable();
        deleteById();
        //updateById();
    }
   
    /**
     * 查询数据
     */
    public static void queryTable(){
        List<Map<String, Object>> list = new ArrayList<>();
        String sql = "select * from user_info";
        Connection connection = getConn();
        try {
            Statement statement = connection.createStatement();
            ResultSet rs  = statement.executeQuery(sql);
            ResultSetMetaData rsmd = rs.getMetaData();
            while(rs.next()){
                Map<String, Object> row = new HashMap<>();
                for(int i = 1; i <= rsmd.getColumnCount(); i++){
                    row.put(rsmd.getColumnName(i), rs.getObject(rsmd.getColumnName(i)));
                }
                list.add(row);
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
   
        //在此可以根据实际需求将解析的数据封装到对象中
        list.stream().forEach(item ->{
            Map<String, Object> rowData = item;
            System.out.println(rowData);
        });
        //System.out.println(list);
    }
   
    /**
     * 创建表
     * @throws Exception
     */
    public static void createTable(String tableSql) throws Exception{
        /*tableSql = "create table t_order02(\n" +
                " id UInt32,\n" +
                " sku_id String,\n" +
                " total_amount Decimal(16,2),\n" +
                " create_time Datetime\n" +
                ") engine =MergeTree\n" +
                " partition by toYYYYMMDD(create_time)\n" +
                " primary key (id)\n" +
                " order by (id,sku_id);";*/
        Connection connection = getConn();
        Statement statement = connection.createStatement();
        boolean execute = statement.execute(tableSql);
        if(execute){
            System.out.println(execute);
            System.out.println("创建表成功");
        }
    }
   
    /**
     * 删除表
     * @throws Exception
     */
    public static void dropTable() throws Exception{
        Connection connection = getConn();
        Statement statement = connection.createStatement();
        statement.execute("drop table t_order01;");
        System.out.println("删除表成功");
    }
   
    /**
     * 插入数据
     * 实际使用时候，插入的语句里面的参数从外部传入进去
     * @throws Exception
     */
    public static void insertOne() throws Exception{
        Connection connection = getConn();
        PreparedStatement pstmt = connection.prepareStatement("insert into t_order01 values('103', 'sku_004', '2500.00','2021-06-01 12:00:00')");
        pstmt.execute();
        System.out.println("insert success");
    }
   
    /**
     * 删除数据
     * 实际使用时候，删除的语句里面的参数从外部传入进去
     */
    public static void deleteById() throws Exception{
        Connection connection = getConn();
        //sku_id ='sku_001'
        PreparedStatement pstmt = connection.prepareStatement("alter table t_order01 delete where sku_id = 'sku_002';");
        pstmt.execute();
        System.out.println("delete success");
    }
   
    /**
     * 修改数据
     * 实际使用时候，修改的语句里面的参数从外部传入进去
     */
    public static void updateById() throws Exception{
        Connection connection = getConn();
        PreparedStatement pstmt = connection.prepareStatement("alter table t_order01 update total_amount=toDecimal32(2000.00,2) where id = '102'");
        pstmt.execute();
        System.out.println("update success");
    }
   
    public static Connection getConn() {
        ClickHouseProperties properties = new ClickHouseProperties();
        properties.setUser(username);
        properties.setPassword(password);
        properties.setDatabase(db);
        properties.setSocketTimeout(socketTimeout);
        ClickHouseDataSource clickHouseDataSource = new ClickHouseDataSource(address, properties);
        ClickHouseConnection conn = null;
        try {
            conn = clickHouseDataSource.getConnection();
            System.out.println(conn);
            System.out.println("连接成功");
            return conn;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }
}

或操作util工具类

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import net.sf.json.JSONObject;
import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.settings.ClickHouseProperties;
  
import java.sql.*;
import java.util.*;
  
/**
 * @Description:
 * @Date 2018/11/12
 */
@Slf4j
@Component
public class ClickHouseUtil {
  
    private static String clickhouseAddress;
  
    private static String clickhouseUsername;
  
    private static String clickhousePassword;
  
    private static String clickhouseDB;
  
    private static Integer clickhouseSocketTimeout;
  
    @Value("${clickhouse.address}")
    public  void setClickhouseAddress(String address) {
        ClickHouseUtil.clickhouseAddress = address;
    }
    @Value("${clickhouse.username}")
    public  void setClickhouseUsername(String username) {
        ClickHouseUtil.clickhouseUsername = username;
    }
    @Value("${clickhouse.password}")
    public  void setClickhousePassword(String password) {
        ClickHouseUtil.clickhousePassword = password;
    }
    @Value("${clickhouse.db}")
    public  void setClickhouseDB(String db) {
        ClickHouseUtil.clickhouseDB = db;
    }
    @Value("${clickhouse.socketTimeout}")
    public  void setClickhouseSocketTimeout(Integer socketTimeout) {
        ClickHouseUtil.clickhouseSocketTimeout = socketTimeout;
    }
  
  
    public static Connection getConn() {
  
        ClickHouseConnection conn = null;
        ClickHouseProperties properties = new ClickHouseProperties();
        properties.setUser(clickhouseUsername);
        properties.setPassword(clickhousePassword);
        properties.setDatabase(clickhouseDB);
        properties.setSocketTimeout(clickhouseSocketTimeout);
        ClickHouseDataSource clickHouseDataSource = new ClickHouseDataSource(clickhouseAddress,properties);
        try {
            conn = clickHouseDataSource.getConnection();
            return conn;
        } catch (SQLException e) {
            e.printStackTrace();
        }
  
        return null;
    }
  
    public static List<JSONObject> exeSql(String sql){
        log.info("cliockhouse 执行sql：" + sql);
        Connection connection = getConn();
        try {
            Statement statement = connection.createStatement();
            ResultSet results = statement.executeQuery(sql);
            ResultSetMetaData rsmd = results.getMetaData();
            List<JSONObject> list = new ArrayList();
            while(results.next()){
                JSONObject row = new JSONObject();
                for(int i = 1;i<=rsmd.getColumnCount();i++){
                    row.put(rsmd.getColumnName(i),results.getString(rsmd.getColumnName(i)));
                }
                list.add(row);
            }
  
            return list;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }
  
}

Test简单使用执行sql查询数据

import com.renrenche.databus.common.ClickHouseUtil;
import com.renrenche.databus.common.Result;
import com.renrenche.databus.domain.logdata.fem.FemParam;
import com.renrenche.databus.service.fem.FemMainService;
import net.sf.json.JSONObject;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
  
import java.util.List;
  
/**
 * @Auther: qixin
 * @Date: 2018/12/11 15:05
 * @Description:
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class SemTest {
  
    @Test
    public void getFrsDataTest(){
        System.out.println("******************");
        String sql="select * from marketing.sem_campaign_real_time_report";
        List<JSONObject> result= ClickHouseUtil.exeSql(sql);
        System.out.println("******************");
    }
  
}

springboot的整合

1、准备一张表，以及表中插入一些实验数据

CREATE TABLE user_info (
  `id` UInt64,
  `user_name` String,
  `pass_word` String,
  `phone` String,
  `create_day` Date DEFAULT CAST(now(),'Date')
)ENGINE = MergeTree
primary key (id)
order by (id);
   
   
INSERT INTO user_info
  (id,user_name,pass_word,phone)
VALUES
  (1,'xiaowang','123456','13325511231'),
  (2,'xiaoma','123456','13825511231'),
  (3,'xiaozhao','123456','18925511231');

2、代码完整整合步骤

导入完整依赖

<dependencies>
   
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
   
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
   
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
   
    <dependency>
        <groupId>org.mybatis.spring.boot</groupId>
        <artifactId>mybatis-spring-boot-starter</artifactId>
        <version>1.3.2</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>5.1.38</version>
    </dependency>
   
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid-spring-boot-starter</artifactId>
        <version>1.1.13</version>
    </dependency>
   
    <dependency>
        <groupId>commons-lang</groupId>
        <artifactId>commons-lang</artifactId>
        <version>2.6</version>
    </dependency>
   
    <!-- clickHouse数据库 -->
    <dependency>
        <groupId>ru.yandex.clickhouse</groupId>
        <artifactId>clickhouse-jdbc</artifactId>
        <version>0.1.53</version>
    </dependency>
   
</dependencies>

2、基础配置文件

server:
  port: 7010
   
  # mybatis 配置
mybatis:
  type-aliases-package: com.congge.entity
  mapper-locations: classpath:/mapper/*.xml
   
spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    click:
      driverClassName: ru.yandex.clickhouse.ClickHouseDriver
      url: jdbc:clickhouse://IP地址:8123/default
      username: default
      password: 123456
      initialSize: 10
      maxActive: 100
      minIdle: 10
      maxWait: 6000

3、使用一个配置类，关联第二步中的click配置属性

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
   
@Component
@ConfigurationProperties(prefix = "spring.datasource.click")
public class ConnectionParamConfig {
   
    private String driverClassName ;
    private String url ;
    private Integer initialSize ;
    private Integer maxActive ;
    private Integer minIdle ;
    private Integer maxWait ;
   
    private String username;
    private String password;
   
    public String getDriverClassName() {
        return driverClassName;
    }
    public void setDriverClassName(String driverClassName) {
        this.driverClassName = driverClassName;
    }
    public String getUrl() {
        return url;
    }
    public void setUrl(String url) {
        this.url = url;
    }
    public Integer getInitialSize() {
        return initialSize;
    }
    public void setInitialSize(Integer initialSize) {
        this.initialSize = initialSize;
    }
    public Integer getMaxActive() {
        return maxActive;
    }
    public void setMaxActive(Integer maxActive) {
        this.maxActive = maxActive;
    }
    public Integer getMinIdle() {
        return minIdle;
    }
    public void setMinIdle(Integer minIdle) {
        this.minIdle = minIdle;
    }
    public Integer getMaxWait() {
        return maxWait;
    }
    public void setMaxWait(Integer maxWait) {
        this.maxWait = maxWait;
    }
   
    public String getUsername() {
        return username;
    }
   
    public void setUsername(String username) {
        this.username = username;
    }
   
    public String getPassword() {
        return password;
    }
   
    public void setPassword(String password) {
        this.password = password;
    }
}

4、重写datasource的配置，使用自定义的clickhouse的属性配置

import javax.annotation.Resource;
   
import com.alibaba.druid.pool.DruidDataSource;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
   
import javax.sql.DataSource;
   
@Configuration
public class DruidConfig {
   
    @Resource
    private ConnectionParamConfig jdbcParamConfig;
   
    /**
     * 重写 DataSource
     * @return
     */
    @Bean
    public DataSource dataSource() {
        DruidDataSource datasource = new DruidDataSource();
        datasource.setUrl(jdbcParamConfig.getUrl());
        datasource.setDriverClassName(jdbcParamConfig.getDriverClassName());
        datasource.setInitialSize(jdbcParamConfig.getInitialSize());
        datasource.setMinIdle(jdbcParamConfig.getMinIdle());
        datasource.setMaxActive(jdbcParamConfig.getMaxActive());
        datasource.setMaxWait(jdbcParamConfig.getMaxWait());
        datasource.setUsername(jdbcParamConfig.getUsername());
        datasource.setPassword(jdbcParamConfig.getPassword());
        return datasource;
    }
   
}

5、提供一个接口和mybatis的查询xml文件

public interface UserInfoMapper {
    void saveData (UserInfo userInfo) ;
    UserInfo selectById (@Param("id") Integer id) ;
    List<UserInfo> selectList () ;
}
   
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.congge.mapper.UserInfoMapper">
    <resultMap id="BaseResultMap" type="com.congge.entity.UserInfo">
        <id column="id" jdbcType="INTEGER" property="id" />
        <result column="user_name" jdbcType="VARCHAR" property="userName" />
        <result column="pass_word" jdbcType="VARCHAR" property="passWord" />
        <result column="phone" jdbcType="VARCHAR" property="phone" />
        <result column="create_day" jdbcType="VARCHAR" property="createDay" />
    </resultMap>
   
    <sql id="Base_Column_List">
        id,user_name,pass_word,phone,create_day
    </sql>
   
    <insert id="saveData" parameterType="com.congge.entity.UserInfo" >
        INSERT INTO user_info
        (id,user_name,pass_word,phone,create_day)
        VALUES
        (#{id,jdbcType=INTEGER},#{userName,jdbcType=VARCHAR},#{passWord,jdbcType=VARCHAR},
        #{phone,jdbcType=VARCHAR},#{createDay,jdbcType=VARCHAR})
    </insert>
   
    <select id="selectById" parameterType="java.lang.Integer" resultMap="BaseResultMap">
        select
        <include refid="Base_Column_List" />
        from user_info
        where id = #{id,jdbcType=INTEGER}
    </select>
   
    <select id="selectList" resultMap="BaseResultMap" >
        select
        <include refid="Base_Column_List" />
        from user_info
    </select>
   
</mapper>

6、接口和服务实现类

@RestController
public class UserInfoController {
   
    @Resource
    private UserInfoService userInfoService ;
   
    //localhost:7010/saveData
    @GetMapping("/saveData")
    public String saveData (){
        UserInfo userInfo = new UserInfo () ;
        userInfo.setId(4);
        userInfo.setUserName("xiaolin");
        userInfo.setPassWord("54321");
        userInfo.setPhone("18500909876");
        userInfo.setCreateDay("2022-02-06");
        userInfoService.saveData(userInfo);
        return "success";
    }
   
    //localhost:7010/getById?id=1
    @GetMapping("/getById")
    public UserInfo getById (int id) {
        return userInfoService.selectById(id) ;
    }
   
    @GetMapping("/getList")
    public List<UserInfo> getList () {
        return userInfoService.selectList() ;
    }
   
}

7、服务实现类，UserInfoService

@Service
public class UserInfoService {
   
    @Resource
    private UserInfoMapper userInfoMapper ;
   
    public void saveData(UserInfo userInfo) {
        userInfoMapper.saveData(userInfo);
    }
   
    public UserInfo selectById(Integer id) {
        return userInfoMapper.selectById(id);
    }
   
    public List<UserInfo> selectList() {
        return userInfoMapper.selectList();
    }
   
}

8、启动类

@SpringBootApplication
@MapperScan(basePackages = {"com.congge.mapper"})
public class App {
   
    public static void main(String[] args) {
        SpringApplication.run(App.class,args);
    }
   
}

9、功能接口测试

查询测试，调用接口：localhost:7010/getById?id=1

然后再去clickhouse表中查询下数据