ClickHouse集群部署与分布式表引擎实战指南

ClickHouse分布式部署、表创建及数据迁移全攻略‌

2025-06-02

所有,分布式,数据库

ClickHouse分布式部署、表创建及数据迁移全攻略

在大数据时代，ClickHouse 以其高性能 OLAP 查询能力和良好的水平扩展性，逐渐成为海量时序与分析场景的首选数据库。要将 ClickHouse 用于生产环境，往往需要部署分布式集群，实现数据的分片与复制，并针对业务场景设计分布式表结构与数据迁移策略。本文将从分布式架构原理出发，结合代码示例与图解，全面介绍如何完成 ClickHouse 分布式集群的部署、表的创建（含复制表与分布式表）、以及数据迁移的多种手段，帮助你快速掌握 ClickHouse 在生产环境的使用要点。

ClickHouse 分布式架构概述
1.1. 单节点 vs 分布式
1.2. 分片（Shard）与副本（Replica）
1.3. ZooKeeper 在分布式中的作用
环境准备与组件安装
2.1. 系统与网络要求
2.2. 安装 ZooKeeper 集群
2.3. 安装 ClickHouse 节点
分布式集群部署示例
3.1. 集群拓扑设计与图解
3.2. ZooKeeper 配置
3.3. ClickHouse config.xml 与 users.xml 配置
3.4. 启动 ClickHouse 服务与校验
分布式表引擎与表创建
4.1. MergeTree 与 ReplicatedMergeTree 引擎
4.2. Distributed 引擎原理与实现
4.3. 本地表与分布式表创建示例
4.4. 示例：查询分布式表的执行流程图解
数据写入、查询与负载均衡
5.1. 写入到 ReplicatedMergeTree 且分片自动路由
5.2. 分布式表查询流程详解
5.3. Insert、Select 示例
数据迁移与同步策略
6.1. 单机 ClickHouse 到分布式集群迁移
6.2. MySQL 到 ClickHouse 的迁移示例（使用 Kafka 或 clickhouse-mysql）
6.3. clickhouse-copier 工具使用
6.4. INSERT SELECT 与外部表引擎同步
6.5. 实时同步示例：使用 Kafka 引擎 + Materialized View
运维与监控要点
7.1. ZooKeeper 集群监控
7.2. ClickHouse 节点健康检查
7.3. 分片与副本恢复流程
7.4. 备份与恢复策略
常见问题与优化建议
8.1. 查询慢或分布式 JOIN 性能优化
8.2. 数据倾斜与分片键设计
8.3. 磁盘、内存、网络调优
总结

1. ClickHouse 分布式架构概述

在深入部署细节之前，首先要明确 ClickHouse 在分布式场景下的几大核心概念：分片（Shard）、副本（Replica）、ZooKeeper 元数据管理，以及分布式表（Distributed Engine）与本地表（MergeTree/ReplicatedMergeTree）的配合。

1.1 单节点 vs 分布式

单节点部署
- 典型用于测试、小规模数据或单机分析。
- 数据存储在本地 MergeTree 或其派生引擎（如 SummingMergeTree、AggregatingMergeTree 等）表中。
- 缺点：无法横向扩展，无副本冗余，节点宕机即数据不可用。
分布式部署
- 通过将数据按某种分片策略均匀分布到多个实例（Shard）上，同时为每个 Shard 配置副本（Replica），实现高可用与水平扩展。
- 查询时，客户端可通过分布式表路由到对应 Shard，或跨 Shard 聚合查询。
- 核心组件：
  - ClickHouse 节点：负责存储与执行。
  - ZooKeeper：负责存储分布式元数据（表的分片 & 副本信息、DDL 同步）。

1.2 分片（Shard）与副本（Replica）

Shard（分片）
- 将逻辑数据集按分片键（如用户 ID、时间范围或哈希值）均匀切分为多个子集，每个子集部署在不同的节点上。
- 常见策略：
  - Hash 分片：shard_key = cityHash64(user_id) % shard_count。
  - 范围分片：根据时间/业务范围拆分。
Replica（副本）
- 每个 Shard 下可部署多个 Replica，保证 Shard 内数据的一致性与高可用。
- Replica 间基于 ZooKeeper 的复制队列自动同步数据。
- 在一个 Replica 挂掉时，点击恢复或重启，其他 Replica 可继续提供服务。

图解：多 Shard / 多 Replica 架构示例

               ┌────────────────────────────────────────────────┐
               │               ZooKeeper 集群（3 节点）          │
               │  存储：/clickhouse/tables/{db}.{table}/shardN   │
               └────────────────────────────────────────────────┘
                      │                   │               │
     ┌────────────────┴─────┐     ┌─────────┴────────┐      │
     │ Shard 1              │     │ Shard 2           │      │
     │ ┌─────────┐ ┌───────┐ │     │ ┌─────────┐ ┌──────┐ │      │
     │ │Replica1 │ │Replica2│ │     │ │Replica1 │ │Replica2│ │      │
     │ │ Node A  │ │ Node B │ │     │ │ Node C  │ │ Node D │ │      │
     │ └─────────┘ └───────┘ │     │ └─────────┘ └──────┘ │      │
     └───────────────────────┘     └─────────────────────┘      │
                      │                   │                   │
                      │                   │                   │
                分布式表路由 / 跨 Shard 聚合查询              │

Shard 内部：Replica1、Replica2 两个副本互为冗余，Replica1、Replica2 分别部署在不同物理机上，以应对单点故障。
跨 Shard：客户端通过分布式表（Distributed Engine）将查询分发至每个 Shard 下的副本，由 ZooKeeper 协调副本选择。

1.3 ZooKeeper 在分布式中的作用

ClickHouse 的分布式功能依赖 ZooKeeper 来保证以下核心功能：

DDL 同步
- 所有 Replica 在创建表、修改表结构时通过 ZooKeeper 写入变更路径，确保各节点同步执行 DDL。
复制队列管理（ReplicatedMergeTree）
- 每个 Replica 会将本地插入/删除任务写入 ZooKeeper 中对应分片的队列节点，其他 Replica 订阅该队列并拉取任务执行，实现数据复制。
分布式表元数据
- Distributed Engine 在 ZooKeeper 中读取集群信息，确定如何将某条 SQL 分发到各个分片。
副本故障检测与恢复
- ZooKeeper 记录当前可用 Replica 列表，当某个 Replica 宕机或网络不可达，其他 Replica 会继续提供写入与查询。

ZooKeeper 目录示例（部分）

/clickhouse/
   ├─ tables/
   │    └─ default.hits/            # hits 表对应的节点
   │         ├─ shard1/             # Shard1 下的所有 Replica
   │         │    ├─ leader_election -> 存储当前 leader 信息
   │         │    └─ queue/Replica1  -> 存储 Replica1 的写入操作
   │         └─ shard2/             # Shard2 下
   │              └─ queue/Replica3
   ├─ macros/                       # 宏定义，可在配置中使用
   └─ replication_alter_columns/... # DDL 同步信息

2. 环境准备与组件安装

本文以 Ubuntu 20.04 为示例操作系统，假设即将部署 2 个 Shard，每个 Shard 2 个 Replica，共 4 台 ClickHouse 节点，并使用 3 节点 ZooKeeper 集群保障高可用。

2.1 系统与网络要求

操作系统
- 建议使用 Debian/Ubuntu/CentOS 等 Linux 发行版，本文以 Ubuntu 20.04 为例。
网络连通性
- 所有节点之间需互相能通：
```
ping zk1 zk2 zk3
ping click1 click2 click3 click4
```
- 关闭防火墙或放通必要端口：
  - ZooKeeper：2181（客户端访问）、2888/3888（集群内部选举）。
  - ClickHouse：9000（TCP 协议，默认客户端端口）、8123（HTTP 接口）、9009（Keeper 通信，若启用 Keeper 模式，可忽略）。
时间同步
- 建议使用 NTP 或 chrony 保证各节点时间同步，否则会影响 ReplicatedMergeTree 的副本选举与健康检查。
```
sudo apt-get install chrony
sudo systemctl enable chrony
sudo systemctl start chrony
```

2.2 安装 ZooKeeper 集群

在 3 台节点（假设 IP 分别为 192.168.1.10/11/12）上完成 ZooKeeper 安装与集群配置。

2.2.1 下载与解压

# 在每台机器执行
wget https://archive.apache.org/dist/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gz
tar -zxvf apache-zookeeper-3.7.1-bin.tar.gz -C /opt/
ln -s /opt/apache-zookeeper-3.7.1-bin /opt/zookeeper

2.2.2 配置 zoo.cfg

# 编辑 /opt/zookeeper/conf/zoo.cfg （如果目录下无 zoo.cfg 示例，可复制 conf/zoo_sample.cfg）
cat <<EOF > /opt/zookeeper/conf/zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/var/lib/zookeeper
clientPort=2181
# 集群内部通信端口（选举与同步）
server.1=192.168.1.10:2888:3888
server.2=192.168.1.11:2888:3888
server.3=192.168.1.12:2888:3888
EOF

2.2.3 创建 dataDir 与 myid

# 在每台机器分别执行
sudo mkdir -p /var/lib/zookeeper
sudo chown $(whoami):$(whoami) /var/lib/zookeeper

# 将编号写入 myid（与 zoo.cfg 中 server.N 对应）
# 机器 192.168.1.10
echo "1" > /var/lib/zookeeper/myid
# 机器 192.168.1.11
echo "2" > /var/lib/zookeeper/myid
# 机器 192.168.1.12
echo "3" > /var/lib/zookeeper/myid

2.2.4 启动 ZooKeeper

# 同步在 3 台节点上启动
/opt/zookeeper/bin/zkServer.sh start
# 检查集群状态
/opt/zookeeper/bin/zkServer.sh status
# 期望输出类似 “Mode: leader” 或 “Mode: follower”

至此，3 节点 ZooKeeper 集群已启动并形成仲裁，可支持多副本 ClickHouse 的元数据管理。

2.3 安装 ClickHouse 节点

在 4 台 ClickHouse 节点（假设 IP 为 192.168.1.20/21/22/23）上，按照以下步骤安装 ClickHouse：

2.3.1 安装 Yandex 官方仓库并安装

# 安装官方 GPG Key
sudo apt-key adv --keyserver keyserver.ubuntu.com --recv E0C56BD4
# 添加仓库
echo "deb https://repo.clickhouse.com/deb/stable/ main/" | sudo tee /etc/apt/sources.list.d/clickhouse.list
# 更新并安装
sudo apt-get update
sudo apt-get install -y clickhouse-server clickhouse-client

2.3.2 配置防火墙与端口

# 放通 TCP 9000、8123、9009 端口（若使用 CentOS，可用 firewalld 或 iptables）
sudo ufw allow 9000/tcp
sudo ufw allow 8123/tcp
sudo ufw allow 9009/tcp
sudo ufw reload

2.3.3 启动 ClickHouse 服务

sudo systemctl enable clickhouse-server
sudo systemctl start clickhouse-server
# 查看日志，确认正常启动
sudo journalctl -u clickhouse-server -f

注意：此时 ClickHouse 还未配置分布式功能，仅是默认的单节点模式。

3. 分布式集群部署示例

下面以 2 Shard × 2 Replica 为例，演示如何将 4 个 ClickHouse 节点组成分布式集群。假设对应节点如下：

Shard1
- Replica1：192.168.1.20（click1）
- Replica2：192.168.1.21（click2）
Shard2
- Replica1：192.168.1.22（click3）
- Replica2：192.168.1.23（click4）

3.1 集群拓扑设计与图解

            ┌────────────────────────────────────────────────┐
            │                   ZooKeeper 3 节点           │
            │   [192.168.1.10, 11, 12] 端口 2181,2888,3888  │
            └────────────────────────────────────────────────┘
                        │              │              │
       ┌────────────────┴──────────────┴──────────────┴───────────────┐
       │                    ClickHouse 分布式集群                       │
       │ Shard1                                  Shard2                 │
       │ ┌───────────┐ ┌───────────┐         ┌───────────┐ ┌───────────┐ │
       │ │ click1    │ │ click2    │         │ click3    │ │ click4    │ │
       │ │ (Replica) │ │ (Replica) │         │ (Replica) │ │ (Replica) │ │
       │ │ zk:2181   │ │ zk:2181   │         │ zk:2181   │ │ zk:2181   │ │
       │ └───────────┘ └───────────┘         └───────────┘ └───────────┘ │
       └───────────────────────────────────────────────────────────────┘
               │                  │              │                  │
               │  ReplicatedMergeTree 本地表 (pathy)  │ Distributed 表 (path) │
               │  数据分片 & 自动复制                 │ 跨 Shard 查询路由     │

ZooKeeper：运行在 192.168.1.10/11/12:2181
click1/click2：Shard1 的 2 个 Replica，两个节点负责存储 Shard1 的数据，数据通过 ZooKeeper 自动复制。
click3/click4：Shard2 的 2 个 Replica，同理。

3.2 ZooKeeper 配置

上文已完成 ZooKeeper 集群搭建，确认集群健康后，ClickHouse 参考以下 ZooKeeper 连接方式即可。

<!-- /etc/clickhouse-server/config.xml （各节点相同，只需保证 zk 配置正确） -->
<yandex>
    <!-- 其他配置省略 -->
    <zookeeper>
        <node>
            <host>192.168.1.10</host>
            <port>2181</port>
        </node>
        <node>
            <host>192.168.1.11</host>
            <port>2181</port>
        </node>
        <node>
            <host>192.168.1.12</host>
            <port>2181</port>
        </node>
    </zookeeper>
    <!-- 更多配置... -->
</yandex>

3.3 ClickHouse `config.xml` 与 `users.xml` 配置

为了实现 ReplicatedMergeTree 与 Distributed 引擎，需修改以下配置文件。

3.3.1 修改 `config.xml`

编辑 /etc/clickhouse-server/config.xml，在 <yandex> 节点内添加以下段落：

<yandex>
    <!-- ... 原有配置 ... -->

    <!-- ZooKeeper 节点 (已如上所示) -->
    <zookeeper>
        <node>
            <host>192.168.1.10</host>
            <port>2181</port>
        </node>
        <node>
            <host>192.168.1.11</host>
            <port>2181</port>
        </node>
        <node>
            <host>192.168.1.12</host>
            <port>2181</port>
        </node>
    </zookeeper>

    <!-- 为分布式部署添加 shards 与 replicas 定义 -->
    <remote_servers>
        <!-- 定义一个逻辑集群名 cluster1，包含 2 个 shard -->
        <cluster1>
            <shard>
                <replica>
                    <host>192.168.1.20</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>192.168.1.21</host>
                    <port>9000</port>
                </replica>
            </shard>
            <shard>
                <replica>
                    <host>192.168.1.22</host>
                    <port>9000</port>
                </replica>
                <replica>
                    <host>192.168.1.23</host>
                    <port>9000</port>
                </replica>
            </shard>
        </cluster1>
    </remote_servers>

    <!-- 定义默认数据库 macros，方便在 SQL 中使用 {cluster} -->
    <macros>
        <cluster>cluster1</cluster>
        <shard>shard1</shard> <!-- 可留空，主要使用 macros.cluster -->
    </macros>

    <!-- 持久化参数，以及其他可选配置 -->
    <!-- ... -->
</yandex>

<remote_servers>
- 定义逻辑集群名称 cluster1，下有两个 <shard> 节点，每个 <shard> 下有若干 <replica>。
- 在后续创建 Distributed 表时，会引用 cluster1，ClickHouse 自动根据此配置将查询分发到各 shard 下的一个副本。
<macros>
- 定义了 {cluster} 宏，后续 SQL 可直接使用 remote('cluster1', ...) 或 {cluster}。

修改完成后，重启 ClickHouse 节点以使配置生效：

sudo systemctl restart clickhouse-server

3.3.2 修改 `users.xml`（可选）

若需为分布式表访问设置白名单，建议修改 /etc/clickhouse-server/users.xml，在相应用户下添加 <networks>：

<!-- users.xml 片段 -->
<profiles>
    <default>
        <!-- 其他配置 -->
    </default>
</profiles>

<users>
    <default>
        <password></password>
        <networks>
            <ip>::/0</ip> <!-- 允许任意 IP 访问 -->
        </networks>
        <profile>default</profile>
        <quota>default</quota>
    </default>
</users>

若公司内部有统一授权管理，可为特定用户专门配置分布式访问权限。

3.4 启动 ClickHouse 服务与校验

重启所有 ClickHouse 节点

sudo systemctl restart clickhouse-server

校验 ZooKeeper 连接
```
clickhouse-client --query="SELECT * FROM system.zookeeper WHERE path LIKE '/clickhouse/%' LIMIT 5;"
```
- 若能正常返回节点信息，则表明 ClickHouse 成功连接到 ZooKeeper。
校验 remote_servers 配置是否生效
在任意一台节点上执行：
```
clickhouse-client --query="SELECT host_name(), version();"
# 查看本地信息
```
然后执行跨集群的 Hello 查询：
```
clickhouse-client --query="SELECT * FROM remote('cluster1', system.one) LIMIT 4;"
```
- 该查询会在 cluster1 下的每个 Replica 上执行 SELECT * FROM system.one LIMIT 1，汇总 4 条记录。如果能正常返回 4 条，则表示 remote\_servers 生效。

4. 分布式表引擎与表创建

在完成分布式部署后，需要了解 ClickHouse 提供的几种常见表引擎，并结合分布式场景设计合适的表结构。

4.1 MergeTree 与 ReplicatedMergeTree 引擎

MergeTree 系列
- 最常用的引擎，适用于单机场景或非严格高可用需求。
- 支持分区（PARTITION BY）、排序键（ORDER BY）、TTL、物化视图等。
- 示例创建：
```
CREATE TABLE default.events_mt (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);
```

ReplicatedMergeTree 系列

在 MergeTree 基础上，增加了通过 ZooKeeper 实现副本复制与容灾能力。
需要传入两个重要参数：
1. ZooKeeper 路径：例如 /clickhouse/tables/{database}.{table}/shardN。
2. Replica 名称：在同一 Shard 下需唯一，如 replica1、replica2。

示例创建（在 Shard1 下的两个 Replica 分别执行）：

CREATE TABLE default.events_shard1_replica1 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard1',  -- ZooKeeper 路径
  'replica1'                                   -- Replica 名称
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

CREATE TABLE default.events_shard1_replica2 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard1',  -- 与 replica1 相同的路径
  'replica2'
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

Shard2 下分别创建两个 Replica

CREATE TABLE default.events_shard2_replica1 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard2',
  'replica1'
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

CREATE TABLE default.events_shard2_replica2 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard2',
  'replica2'
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

说明：
ZooKeeper 路径 '/clickhouse/tables/default.events/shard1' 与 Shard 名称保持一致，有助于后续维护。
每个 Shard 下的 Replica 都指定相同的 ZooKeeper 路径，Replica 在同一路径上协调数据复制。

4.2 Distributed 引擎原理与实现

Distributed 引擎
- 提供跨 Shard 的查询路由能力，本质上是一个逻辑视图，将查询分发到建在各 Shard 下的本地表，再在客户端聚合结果。
- 创建时需要指定：
  1. 集群名称：与 config.xml 中 remote_servers 配置保持一致，如 cluster1。
  2. 数据库和表名：在各 Replica 上实际存在的本地表名（如 default.events_shard1_replica1、..._replica2、...shard2_replica1、...shard2_replica2）。
  3. 分片键（可选）：用于将写入分发到某个 Shard，而不是广播到所有 Shard。
- 示例创建：
```
CREATE TABLE default.events_distributed (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = Distributed(
  'cluster1',    -- 与 config.xml 中 remote_servers 的 <cluster1>
  'default',     -- 数据库名
  'events_local',-- 各 Shard 对应的本地表前缀（需在各节点上创建同名本地表）
  rand()         -- 分片键，可改为 cityHash64(user_id)
);
```
- 由于各 Shard 下的本地表可能使用 ReplicatedMergeTree 并加入了 Replica 后缀，为简化管理，可在各 local 表下创建一个同名别名表 events_local，指向当前 Replica。示例：
  每台节点（click1\~click4）都创建一个同名的本地别名表：
```
CREATE TABLE default.events_local AS default.events_shard1_replica1;  -- click1
CREATE TABLE default.events_local AS default.events_shard1_replica2;  -- click2
CREATE TABLE default.events_local AS default.events_shard2_replica1;  -- click3
CREATE TABLE default.events_local AS default.events_shard2_replica2;  -- click4
```
  这样，在 Distributed 引擎中只需引用 events_local，ClickHouse 会自动查找每个节点上对应的本地表。

4.3 本地表与分布式表创建示例

下面结合 Shard1/Shard2、Replica1/Replica2 全流程示例。

4.3.1 Shard1 Replica1 上创建本地表

-- 点击 click1 (Shard1 Replica1)
CREATE DATABASE IF NOT EXISTS default;

CREATE TABLE default.events_shard1_replica1 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard1',
  'replica1'
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

4.3.2 Shard1 Replica2 上创建本地表

-- 点击 click2 (Shard1 Replica2)
CREATE DATABASE IF NOT EXISTS default;

CREATE TABLE default.events_shard1_replica2 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard1',
  'replica2'
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

4.3.3 Shard2 Replica1 上创建本地表

-- 点击 click3 (Shard2 Replica1)
CREATE DATABASE IF NOT EXISTS default;

CREATE TABLE default.events_shard2_replica1 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard2',
  'replica1'
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

4.3.4 Shard2 Replica2 上创建本地表

-- 点击 click4 (Shard2 Replica2)
CREATE DATABASE IF NOT EXISTS default;

CREATE TABLE default.events_shard2_replica2 (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = ReplicatedMergeTree(
  '/clickhouse/tables/default.events/shard2',
  'replica2'
)
PARTITION BY toYYYYMM(dt)
ORDER BY (user_id, dt);

提示：在创建完上述本地表后，可使用以下命令检查副本同步是否正常：
-- 在任意节点执行
SELECT
  database,
  table,
  is_leader,
  queue_size,
  future_parts,
  parts_to_merge,
  last_queue_update,
  last_queue_update_time
FROM system.replicas
WHERE database = 'default' AND table LIKE 'events%';
查看 is_leader、queue_size 是否为 0，表示副本同步正常；若有积压任务，可等待或手动修复。

4.3.5 在每个节点上创建本地别名表

为了让分布式引擎统一使用同名本地表，建议在每个节点上都创建一个 events_local 别名表，指向上一步创建的 Replica 表。示例如下：

click1（Shard1 Replica1）

CREATE TABLE default.events_local AS default.events_shard1_replica1;

click2（Shard1 Replica2）

CREATE TABLE default.events_local AS default.events_shard1_replica2;

click3（Shard2 Replica1）

CREATE TABLE default.events_local AS default.events_shard2_replica1;

click4（Shard2 Replica2）

CREATE TABLE default.events_local AS default.events_shard2_replica2;

说明：别名表不会在存储目录再新建数据；它只是一个对 ReplicatedMergeTree 本地表的引用（ATTACH TABLE 方式）。如果希望更严格隔离，也可以使用 ATTACH TABLE 语法，但 AS ... 方式足够常见。

4.3.6 创建分布式表

在任意一台节点（建议使用 click1）上执行：

CREATE TABLE default.events_distributed (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = Distributed(
  'cluster1',         -- 与 config.xml 中定义的集群名称
  'default',          -- 数据库名
  'events_local',     -- 各节点上本地表别名
  cityHash64(user_id) -- 分片键
);

关键说明：
cityHash64(user_id)：ClickHouse 内置的一种哈希函数，可将 user_id 映射到 [0, 2^64) 区间后再 % shard_count，分散写入到不同的 Shard。
如果不填分片键（如填 rand() 或 ''），则 Insert 操作会自动将每条记录广播到所有 Shard。

到此，分布式表与本地 Replica 表的创建已完成。

4.4 示例：查询分布式表的执行流程图解

┌─────────────────────────────────────────────────────────────────────────┐
│                         ClickHouse Client                              │
│   SELECT user_id, count() FROM default.events_distributed GROUP BY user_id  │
└─────────────────────────────────────────────────────────────────────────┘
                             │
                   查询路由到 cluster1
                             │
        ┌────────────────────┴────────────────────┐
        │                                         │
┌───────────────┐                       ┌───────────────┐
│    Shard1     │                       │    Shard2     │
│ (click1/2)    │                       │ (click3/4)    │
│ Distributed   │                       │ Distributed   │
│ Engine Worker │                       │ Engine Worker │
└───────┬───────┘                       └───────┬───────┘
        │      查询对应本地表 events_local             │      查询对应本地表 events_local
        ▼                                         ▼
┌───────────────┐                       ┌───────────────┐
│ Local Table   │                       │ Local Table   │
│ events_local  │                       │ events_local  │
│ (Shard1 Data) │                       │ (Shard2 Data) │
│ ReplicatedMT  │                       │ ReplicatedMT  │
└───────┬───────┘                       └───────┬───────┘
        │                                         │
        │ 执行 group by、count() 本地聚合            │ 执行本地聚合
        │                                         │
        ▼                                         ▼
┌──────────────────┐                     ┌──────────────────┐
│ Partial Results  │                     │ Partial Results  │
│ (user_id, count) │                     │ (user_id, count) │
└──────────┬───────┘                     └──────────┬───────┘
           │                                         │
           │         将部分结果汇总到客户端并进行最终合并         │
           └───────────────┬─────────────────────────────────────┘
                           ▼
                    客户端合并聚合结果
                           │
                           ▼
               返回最终 (user_id, total_count) 列表

Shard1/Shard2：分布式表引擎仅充当调度者，真正的计算在各节点本地的 events_local。
本地聚合：为了减少网络传输，ClickHouse 默认会先在本地执行 GroupBy、聚合等操作，只有聚合后较小的中间结果通过网络返回再做最终合并。这样能显著提高分布式查询性能。

5. 数据写入、查询与负载均衡

完成表结构创建后，接下来演示如何将数据写入分布式表与查询，以及写入时如何自动分片或广播。

5.1 写入到 ReplicatedMergeTree 且分片自动路由

使用分布式表写入
- 推荐通过分布式表 events_distributed 写入，ClickHouse 会根据 cityHash64(user_id) % shard_count 自动将数据路由到相应 Shard 的 Replica（随机选择一个可用 Replica 写入）。
- 示例插入 3 条数据，user\_id 为 1、2、3：
```
INSERT INTO default.events_distributed VALUES
('2023-09-01', 1, 'click', 10.5),
('2023-09-01', 2, 'view', 5.0),
('2023-09-01', 3, 'purchase', 100.0);
```
  - 若 Shard Count=2，那么：
    - 对于 user_id = 1：cityHash64(1) % 2 = 1（假设），路由到 Shard2；
    - user_id = 2：%2 = 0，写入 Shard1；
    - user_id = 3：%2 = 1，写入 Shard2。
写入副本选择
- Shard 内部多个 Replica 会随机选择一个可写 Replica；若写入的 Replica 挂掉，其他 Replica 会接受写入请求。写入后，Replica 间基于 ZooKeeper 自动同步数据。

5.2 分布式表查询流程详解

查询 events_distributed
- 当执行 SELECT * FROM events_distributed WHERE user_id = 2; 时，ClickHouse 会根据分片键 cityHash64(2) % 2 计算出目标 Shard（Shard1），并将查询请求发给 Shard1 的一个 Replica。
- 然后在该 Replica 上查询 events_local（即 Shard1 本地的 ReplicatedMergeTree 表），返回结果。
- 如果 Query 涉及跨 Shard（如 GROUP BY 或不带 WHERE 的 SELECT *），则请求会广播到所有 Shard，每个 Shard 返回部分结果，最后由客户端合并。
分布式聚合与性能
- 对于大表聚合查询，分布式表引擎会首先在每个 Shard 本地进行“部分聚合（partial aggregation）”，然后再把各 Shard 的部分结果收集到一个节点进行“最终聚合（final aggregation）”，大幅减少网络传输量。

5.3 Insert、Select 示例

批量插入示例

INSERT INTO default.events_distributed
SELECT 
  toDate('2023-09-02') AS dt, 
  number AS user_id, 
  'auto' AS action, 
  number * 1.1 AS value
FROM numbers(100000)  -- 生成 100,000 条测试数据
WHERE number < 10000; -- 只写入前 10,000 条

查询示例

-- 查看 Shard1 上的数据量（仅在 Shard1 的 click1 或 click2 节点上执行）
SELECT count(*) FROM default.events_shard1_replica1;
SELECT count(*) FROM default.events_shard1_replica2;

-- 查询分布式表中的总数据量
SELECT count(*) FROM default.events_distributed;

-- 分布式聚合示例
SELECT user_id, count() AS cnt
FROM default.events_distributed
GROUP BY user_id
ORDER BY cnt DESC
LIMIT 10;

验证数据一致性
在 Shard1 Replica1 与 Replica2 上分别查询本地表，确认两者数据同步：
```
SELECT count(*) FROM default.events_shard1_replica1;
SELECT count(*) FROM default.events_shard1_replica2;
```

6. 数据迁移与同步策略

在实际生产中，经常需要将已有数据迁移到新的分布式 ClickHouse 集群，或与外部数据库（如 MySQL）集成，实现实时或离线数据同步。下面介绍几种常见迁移与同步方案。

6.1 单机 ClickHouse 到分布式集群迁移

假设已有一个单节点 ClickHouse（192.168.1.30），其中有表 default.events_single，需要将其数据迁移到上述分布式集群并保持不间断服务。

6.1.1 在新集群创建同结构的分布式表

在新集群创建 ReplicatedMergeTree 本地表与 Distributed 表（与前节示例一致）。
确保 events_distributed 已就绪。

6.1.2 使用 `INSERT SELECT` 迁移数据

在原单节点上执行以下操作，将数据复制到分布式表（通过 clickhouse-client 连接到分布式集群任一节点即可）：

clickhouse-client --host=192.168.1.20 --query="
INSERT INTO default.events_distributed
SELECT * FROM remote('single_host', default, 'events_single')
"

需先在 config.xml 的 remote_servers 中配置 single_host，以便分布式查询原节点数据。示例配置（在每个新集群节点的 /etc/clickhouse-server/config.xml 添加）：

<remote_servers>
    <single_host_cluster>
        <shard>
            <replica>
                <host>192.168.1.30</host>
                <port>9000</port>
            </replica>
        </shard>
    </single_host_cluster>
</remote_servers>

然后在新集群中执行：

INSERT INTO default.events_distributed
SELECT * FROM remote('single_host_cluster', default, 'events_single');

上述操作会将单节点数据分批读取，并插入到分布式表，分布式表会自动分片到各 Shard。在数据量大的情况下，建议拆分范围分批执行，例如按照 dt 范围分区多次执行。

6.1.3 增量同步

在完成初次全量迁移后，可使用 ZooKeeper + Kafka 或持续抓取增量数据进入分布式表，以实现接近实时的迁移。

方案一：Materialized View + Kafka
- 在原单节点 ClickHouse 上创建一个 Kafka 引擎表，订阅写入事件；
- 创建一个 Materialized View，将 Kafka 中的数据插入到新集群的分布式表。
方案二：Debezium + Kafka Connect
- 使用 Debezium 将 MySQL/ClickHouse 的 Binlog 推到 Kafka；
- ClickHouse 侧使用 Kafka 引擎与 Materialized View 实时消费，插入分布式表。

6.2 MySQL 到 ClickHouse 的迁移示例（使用 Kafka 或 `clickhouse-mysql`）

很多场景需要将 MySQL 中的业务表迁移到 ClickHouse 进行高性能 OLAP 查询。常用方案如下：

6.2.1 使用 Kafka + ClickHouse Kafka 引擎

在 MySQL 中开启 Binlog，并使用 Kafka Connect + Debezium 将数据写入 Kafka 主题（如 mysql.events）。

在 ClickHouse 集群上创建 Kafka 引擎表

CREATE TABLE default.events_kafka (
  `dt` Date,
  `user_id` UInt64,
  `action` String,
  `value` Float32
) ENGINE = Kafka SETTINGS
  kafka_broker_list = 'kafka1:9092,kafka2:9092',
  kafka_topic_list = 'mysql.events',
  kafka_group_name = 'ch_consumer_group',
  kafka_format = 'JSONEachRow',
  kafka_num_consumers = 4;

创建 Materialized View
- Materialized View 将消费 events_kafka，并将数据插入分布式表：
```
CREATE MATERIALIZED VIEW default.events_mv TO default.events_distributed AS
SELECT
  dt,
  user_id,
  action,
  value
FROM default.events_kafka;
```
- 这样，Kafka 中的新数据会自动被 MV 推送到分布式表，实现实时同步。

6.2.2 使用 `clickhouse-mysql` 工具

clickhouse-mysql 是社区提供的一个 Python 脚本，可直接将 MySQL 表结构与数据迁移到 ClickHouse。

安装依赖
```
pip install clickhouse-mysql
```

执行迁移命令

clickhouse-mysql --mysql-host mysql_host --mysql-port 3306 --mysql-user root --mysql-password secret \
  --clickhouse-host 192.168.1.20 --clickhouse-port 9000 --clickhouse-user default --clickhouse-password '' \
  --database mydb --table events --clickhouse-database default --clickhouse-table events_distributed

默认会将 MySQL 表自动映射为 ClickHouse 表，如创建合适的 MergeTree 引擎表，再批量插入数据。
对于分布式环境，可先在新集群创建分布式表，再指定 --clickhouse-table 为分布式表，脚本会自动往分布式表写入数据。

6.3 `clickhouse-copier` 工具使用

clickhouse-copier 是 ClickHouse 社区自带的工具，可在集群内部做分片间或集群间的数据搬迁。

准备复制任务的配置文件（copier_config.xml）

<copy>
  <shard>
    <cluster>cluster1</cluster>
    <replica>click1</replica>
  </shard>
  <shard>
    <cluster>cluster1</cluster>
    <replica>click3</replica>
  </shard>

  <tables>
    <table>
      <database>default</database>
      <name>events_local</name>
    </table>
  </tables>
</copy>

上述示例将指定将 events_local 从 Shard1 的 click1 复制到 Shard2 的 click3，需根据实际场景配置更多 <shard> 与 <table>。

执行复制
```
clickhouse-copier --config /path/to/copier_config.xml --replication 0
```
- --replication 0 表示不做 ReplicatedMergeTree 的基于日志复制，仅做一次全量迁移。
- 适用于集群扩容、分片重平衡等操作。

6.4 `INSERT SELECT` 与外部表引擎同步

INSERT SELECT
- 适用于跨集群、跨数据库全量复制：
```
INSERT INTO default.events_distributed
SELECT * FROM default.events_local WHERE dt >= '2023-09-01';
```
- 可分批（按日期、ID 范围）多次执行。
外部表引擎
- ClickHouse 支持通过 MySQL 引擎访问 MySQL 表，如：
```
CREATE TABLE mysql_events (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
)
ENGINE = MySQL('mysql_host:3306', 'mydb', 'events', 'root', 'secret');
```
- 然后可在 ClickHouse 侧做：
```
INSERT INTO default.events_distributed
SELECT * FROM mysql_events;
```
- 外部表引擎适合数据量相对较小或批量一次性导入，若是实时增量同步，仍推荐 Kafka + Materialized View。

6.5 实时同步示例：使用 Kafka 引擎 + Materialized View

在 MySQL 侧将 Binlog 推到 Kafka 后，ClickHouse 侧通过 Kafka 引擎表 + MV，实现近实时同步。

MySQL → Kafka
- 部署 Kafka 集群。
- 使用 Debezium Connector for MySQL，将 MySQL Binlog 写入 Kafka 主题 mysql.events_binlog。

ClickHouse 侧创建 Kafka 表

CREATE TABLE default.events_binlog_kafka (
  dt Date,
  user_id UInt64,
  action String,
  value Float32
) ENGINE = Kafka SETTINGS
  kafka_broker_list = 'k1:9092,k2:9092',
  kafka_topic_list = 'mysql.events_binlog',
  kafka_group_name = 'ch_binlog_consumer',
  kafka_format = 'JSONEachRow',
  kafka_num_consumers = 4;

创建 Materialized View
```
CREATE MATERIALIZED VIEW default.events_binlog_mv TO default.events_distributed AS
SELECT dt, user_id, action, value
FROM default.events_binlog_kafka;
```
- 当 Kafka 有新消息（INSERT/UPDATE/DELETE）时，MV 自动触发，将数据写入分布式表。
- 对于 UPDATE/DELETE，可根据具体业务需求将这些操作转化为 ClickHouse 的 MergeTree 修改或 VXIN 等逻辑。

7. 运维与监控要点

在生产环境下，ClickHouse 分布式集群的健壮性和性能调优尤为关键。以下介绍一些常见的运维与监控要点。

7.1 ZooKeeper 集群监控

节点状态检查

echo ruok | nc 192.168.1.10 2181  # 返回 imok 则正常
echo stat | nc 192.168.1.10 2181  # 查看节点状态、客户端连接数

集群状态检查
```
echo srvr | nc 192.168.1.10 2181
```
- 可查看是否有选举 leader、是否存在掉线节点等。
监控指标
- 使用 Prometheus JMX Exporter + Grafana 监控 ZooKeeper 的请求延时、连接数、选举延时等。

7.2 ClickHouse 节点健康检查

系统表

system.replication_queue：查看各 Replica 的复制队列积压情况。

SELECT database, table, is_currently_executing, parts_to_merge, queue_size 
FROM system.replication_queue;

system.mutations：查看表的 mutations（更新/删除）状态。

SELECT database, table, mutation_id, is_done, parts_to_do, parts_done 
FROM system.mutations;

system.parts：查看数据分区与磁盘占用情况。

SELECT database, table, partition, name, active, bytes_on_disk 
FROM system.parts WHERE database='default' AND table LIKE 'events%';

system.metrics / system.events：监控 ClickHouse 实时指标，如 Query、Insert 吞吐量，Cache 命中率等。

持续监控
- 使用 ClickHouse-Prometheus Exporter + Grafana，可监控：
  - Queries per second
  - Replication lag
  - Merge 任务数
  - 内存使用占比
  - 磁盘 I/O

7.3 分片与副本恢复流程

7.3.1 Replica 加入流程

新增 Replica
- 在一个 Shard 下新增 Replica，先在 ZooKeeper 对应路径下创建新 Replica 的目录。
- 在新节点上创建本地表（表结构需与原 Shard 保持一致），并指定新的 Replica 名称。
- 启动 ClickHouse，该 Replica 会从 ZooKeeper 上的复制队列拉取现有数据，完成全量数据复制。
Shard 扩容（横向扩容）
- 如果要增加 Shard 数量（比如从 2 个 Shard 扩容到 3 个），则需：
  - 暂停写入，或者使用 clickhouse-copier 做分片重平衡。
  - 在新节点上创建对应的本地 ReplicatedMergeTree 表，指定新的 Shard 路径。
  - 使用 clickhouse-copier 或脚本将已有数据重分布到新的 Shard。

7.3.2 副本修复流程

当某个 Replica 节点发生故障并恢复后，需要让它重新同步数据：

重启节点，它会检测到 ZooKeeper 上已有的副本信息。
Replica 恢复复制，从 Leader 主动拉取尚未复制的分区文件并恢复。

检查状态

SELECT database, table, replica_name, is_leader, queue_size 
FROM system.replicas WHERE database='default' AND table LIKE 'events%';

queue_size=0 且 is_currently_executing=0 表示恢复完成。

7.4 备份与恢复策略

备份工具
- Altinity ClickHouse Backup：社区推荐备份工具。支持全量/增量备份与恢复。
- 也可手动使用 clickhouse-client --query="SELECT * FROM table FORMAT Native" 导出，然后再用 clickhouse-client --query="INSERT INTO table FORMAT Native" 导入。
ZooKeeper 数据备份
- 可使用 zkCli.sh 导出关键路径的节点数据，以及定期备份 /var/lib/zookeeper/version-2。
恢复流程
1. 恢复 ZooKeeper 数据，保证 ReplicatedMergeTree 的队列信息完整。
2. 重启 ClickHouse，Replica 会从 ZooKeeper 获取需要恢复的分区；
3. 如果只想恢复部分数据，可手动删除对应的本地分区文件，再让 Replica 重新执行复制。

8. 常见问题与优化建议

在 ClickHouse 分布式生产环境中，经常会遇到性能瓶颈、数据倾斜、Shard 节点不均衡等问题。下面总结一些常见问题与优化技巧。

8.1 查询慢或分布式 JOIN 性能优化

避免跨 Shard JOIN
- ClickHouse 的分布式 JOIN 在多 Shard 场景下需要将数据从一个 Shard 拉取到另一个 Shard 进行 Join，网络 I/O 成本高。建议：
  - 数据预聚合（Denormalization）：将需要关联的数据预先合并到同一个表中；
  - 使用物化视图：在本地 MergeTree 表上预先计算好关键信息；
  - 单 Shard 物理表：如果某个表非常小，可把它复制到每个 Shard 上本地 Join。
Distributed 聚合优化
- 对于大规模聚合查询，建议先在本地执行聚合（aggregate_overflow_mode='throw'），再在客户端进行最终合并。
- 使用 settings max_threads = X, max_memory_usage = Y 控制查询资源消耗。

8.2 数据倾斜与分片键设计

数据倾斜
- 如果分片键导出的数据在某个 Shard 过多而其他 Shard 较少，导致 Shard1 负载过重，Shards2/3 空闲。
- 解决方案：
  - 重新设计分片键，例如使用复合键或哈希函数与随机数结合；
  - 动态调整分片策略，使用一致性哈希等更均衡的方案；
  - 扩容 Shard 节点，将热点数据分摊到更多 Shard。

8.3 磁盘、内存、网络调优

磁盘性能
- 推荐使用 SSD 或 NVMe，至少提供 10,000+ IOPS；
- ClickHouse 在 Merge 任务、高并发写入时对磁盘 I/O 敏感。可使用 RAID0 多盘并行提升吞吐。
内存配置
- 设置合理的 max_memory_usage
- 调整 [max_threads] 来控制并行度，避免 OOM；
- 若有大量 Map/Join 操作，可考虑开启 [join_use_nulls_for_low_cardinality_keys] 以减少内存占用。
网络带宽与延迟
- 分布式查询与复制都依赖网络：
  - 使用至少 10Gb/s 以降低跨 Shard 数据传输延迟；
  - 配置 max_distributed_connections、receive_timeout、send_timeout 等参数优化通信。

9. 总结

本文从 ClickHouse 分布式架构原理入手，详细讲解了如何在生产环境下：

部署 ZooKeeper 高可用集群，并配置 ClickHouse 节点连接；
设计分布式集群拓扑，实现 Shard 与 Replica 的高可用与负载均衡；
在各节点创建 ReplicatedMergeTree 本地表，通过 ZooKeeper 管理副本复制；
使用 Distributed 引擎创建逻辑表，自动实现跨 Shard 路由与分布式聚合；
演示数据写入与查询流程，并提供批量 Insert、Distributed 聚合等常见操作示例；
提供多种数据迁移方案，包括单机→分布式迁移、MySQL→ClickHouse 同步、Kafka 实时同步等全流程；
总结运维与监控要点，探讨 Replica 恢复、Shard 扩容、性能调优等实战经验；
针对常见问题给出优化建议，如数据倾斜、跨 Shard JOIN 降低网络开销、硬件选型等。

通过本文内容，你可以：

搭建一个稳定的 ClickHouse 分布式集群，实现数据的高可用与水平扩展；
利用 ReplicatedMergeTree 与 Distributed 引擎，灵活构建分布式表结构；
结合 Kafka、Materialized View、clickhouse-copier 等工具，实现多源异构数据迁移与实时同步；
在运维过程中通过系统表与监控手段快速排查问题，保证集群高效运行；
通过合理的分片键与硬件优化，避免数据倾斜与性能瓶颈。

实时数据分析数据库 ClickHouse 介绍

2024-09-09

ClickHouse是一个用于在线分析(OLAP)的列式数据库管理系统(DBMS)。它的主要特点是极高的数据处理速度，可以在几十秒内完成对千万条记录的查询。

ClickHouse的主要特性包括：

列式存储：数据按列存储，适合于某些类型的查询操作。
数据压缩：支持多种数据压缩算法，以减少存储空间。
向量化执行引擎：使用SIMD指令优化代码执行。
索引：适度的索引减少了查询时的seek和IO操作。
多核并行处理：利用多核处理器进行并行处理。
SQL支持：支持标准的SQL查询。

安装和使用ClickHouse：

下载并安装：访问ClickHouse官网下载页面，选择合适的版本并安装。
启动服务：在安装目录下执行clickhouse-server启动服务。
连接数据库：使用clickhouse-client连接数据库。

基本SQL操作示例：




-- 创建数据库
CREATE DATABASE IF NOT EXISTS example_db;
 
-- 创建表
CREATE TABLE example_db.example_table (
  EventDate Date,
  EventTime DateTime,
  UserID Int32,
  EventName String,
  EventData String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(EventDate)
ORDER BY (EventDate, EventTime, intHash32(UserID))
 
-- 插入数据
INSERT INTO example_db.example_table (EventDate, EventTime, UserID, EventName, EventData) VALUES ('2023-01-01', '2023-01-01 00:00:00', 42, 'event_name', 'event_data');
 
-- 查询数据
SELECT EventDate, count(UserID) FROM example_db.example_table GROUP BY EventDate;

以上是ClickHouse的基本介绍和使用示例。

Postgresql数据同步到Clickhouse

2024-09-09

为了将PostgreSQL数据同步到ClickHouse，你可以使用外部数据同步工具，例如:

Debezium: 一个分布式平台，用于捕获数据库变更。
Kafka: 一个分布式流平台。
pg\_chameleon: 一个用于PostgreSQL和ClickHouse之间数据同步的工具。

以下是使用pg\_chameleon的一个基本示例：

安装pg\_chameleon：




# 安装依赖
sudo apt-get install -y postgresql-12-replication
 
# 下载pg_chameleon
git clone https://github.com/2ndquadrant-it/pg_chameleon.git /usr/src/pg_chameleon
 
# 编译安装
cd /usr/src/pg_chameleon
make USE_PGXS=1
make USE_PGXS=1 install

配置PostgreSQL和ClickHouse：

确保PostgreSQL有复制权限和对应的数据库已经创建好。

配置pg\_chameleon：

在PostgreSQL中创建一个角色和对应的复制槽，然后在pg\_chameleon中配置连接到PostgreSQL和ClickHouse。

启动pg\_chameleon：




pg_chameleon -D /path/to/config/directory

请注意，这只是一个基本示例，实际部署时可能需要考虑更多因素，如数据同步的实时性、一致性和安全性等。同时，你需要根据你的系统环境和需求调整安装和配置步骤。

ClickHouse的数据导出与接口

2024-09-06

在ClickHouse中，数据可以通过多种方式导出，包括使用命令行工具、SQL查询以及编程语言接口。以下是一个使用Python clickhouse-driver库从ClickHouse数据库导出数据的示例：

首先，确保你已经安装了clickhouse-driver库：




pip install clickhouse-driver

然后，使用Python代码导出数据：




from clickhouse_driver import Client
 
# 创建ClickHouse客户端连接
client = Client('localhost')
 
# 执行SQL查询并获取数据
result = client.execute('SELECT * FROM your_table')
 
# 将数据导出为CSV文件
with open('output.csv', 'w') as file:
    for row in result:
        file.write(','.join(map(str, row)) + '\n')

在这个例子中，your_table是你希望导出数据的表名，代码将该表的所有数据导出到名为output.csv的文件中。你可以根据需要修改查询和导出的文件路径。

2024-09-06

在ClickHouse中，你可以使用system.query_log表来查看正在执行的SQL语句的信息。这个表记录了所有查询的日志，包括查询开始执行的时间、查询执行的时长、查询的状态等。

以下是一个简单的SQL查询，用于获取当前正在执行的所有SQL语句的概要信息：




SELECT
    query_start_time,
    query,
    read_rows,
    total_rows_approx,
    memory_usage,
    state
FROM system.query_log
WHERE is_initial_query AND state = 'Running'

这个查询会返回以下信息：

query_start_time: 查询开始的时间。
query: 查询的SQL语句。
read_rows: 查询执行过程中读取的行数。
total_rows_approx: 查询执行过程中估算的总行数。
memory_usage: 查询执行过程中使用的内存量。
state: 查询的当前状态（例如，'Running'表示查询正在执行）。

请注意，system.query_log表中的数据是有一定时效性的，数据保留的时间由ClickHouse配置文件中的query_log_max_age配置项决定。如果你需要保留更长时间的查询日志，可以考虑增加这个配置项的值。

此外，根据你的需求，你可能还需要根据其他列（如user、address等）来筛选或者排序查询结果。

高级DBA教你clickhouse数据导入MYSQL数据库技术方法

2024-09-06

在实际的生产环境中，我们常常需要将ClickHouse的数据导入到MySQL数据库中。这里提供一种方法，使用ClickHouse的INSERT INTO语句和MySQL的LOAD DATA INFILE语句进行导入。

首先，我们需要在ClickHouse中创建一个导出数据的临时表，假设我们要导出的表名为click_table，并且该表已经存在。




-- 创建临时表
CREATE TEMPORARY TABLE IF NOT EXISTS click_table_tmp AS click_table ENGINE = Log;

然后，我们可以使用INSERT INTO语句将数据导入到临时表中：




-- 清空临时表
DELETE FROM click_table_tmp;
 
-- 从原表插入数据到临时表
INSERT INTO click_table_tmp SELECT * FROM click_table;

接下来，我们需要将临时表中的数据导出到一个CSV文件中，该文件将被MySQL用来导入数据。




-- 导出数据到CSV文件
INSERT INTO TABLE click_table_tmp FORMAT CSVWithNamesAndTypes;

在MySQL中，我们需要创建一个与ClickHouse表结构相同的表，然后使用LOAD DATA INFILE语句将CSV文件导入到MySQL表中。




-- 在MySQL中创建表
CREATE TABLE IF NOT EXISTS mysql_table (
    -- 这里列出所有ClickHouse表中的字段和对应的MySQL数据类型
    column1 INT,
    column2 VARCHAR(255),
    ...
);
 
-- 导入数据
LOAD DATA INFILE '/path/to/click_table.csv'
INTO TABLE mysql_table
FIELDS TERMINATED BY ',' -- 根据CSV文件实际情况调整
ENCLOSED BY '"' -- 如果CSV文件字段被引号包围的话
LINES TERMINATED BY '\n' -- 行结束符，根据操作系统调整
IGNORE 1 LINES; -- 如果CSV文件有标题头，则忽略第一行

请注意，这个过程需要你有足够的权限来执行这些操作，并且在执行这些操作之前需要备份你的数据。此外，CSV文件的路径和格式需要根据你的服务器环境进行相应的调整。

常用ClickHouse性能监控SQL汇总

2024-09-06




-- 查询所有分区的大小
SELECT 
    database, 
    table, 
    partition, 
    sum(bytes_on_disk) AS partition_size 
FROM system.parts 
GROUP BY database, table, partition 
ORDER BY partition_size DESC;
 
-- 查询每个表的大小
SELECT 
    database, 
    table, 
    sum(bytes_on_disk) AS table_size 
FROM system.parts 
GROUP BY database, table 
ORDER BY table_size DESC;
 
-- 查询整个集群的大小
SELECT 
    database, 
    sum(bytes_on_disk) AS database_size 
FROM system.parts 
GROUP BY database 
ORDER BY database_size DESC;
 
-- 查询所有节点的CPU和内存使用情况
SELECT 
    hostname, 
    cpu_usage, 
    memory_usage_percent 
FROM system.metrics 
WHERE (event_date >= today() - 1) AND (event_date <= today());
 
-- 查询最慢的查询
SELECT 
    query, 
    round(elapsed, 2) AS duration, 
    type 
FROM system.query_log 
WHERE (type = 'Query' AND event_date >= today() - 1 AND event_date <= today()) 
ORDER BY duration DESC 
LIMIT 10;
 
-- 查询最频繁的查询
SELECT 
    query, 
    count() AS num_executions 
FROM system.query_log 
WHERE (type = 'Query' AND event_date >= today() - 1 AND event_date <= today()) 
GROUP BY query 
ORDER BY num_executions DESC 
LIMIT 10;

这些SQL语句可以帮助你监控ClickHouse的性能，包括分区大小、表大小、集群大小，以及节点的CPU和内存使用情况。同时，它们还提供了最慢查询和最频繁查询的相关信息。这些信息对于调优ClickHouse集群性能至关重要。

Springboot里集成Mybatis-plus、ClickHouse

2024-09-06

所有,java

Spring Boot 整合 MyBatis-Plus 和 ClickHouse 的方式主要是配置数据源和定义相应的 Mapper。以下是一个简单的示例：

在 pom.xml 中添加依赖：




<!-- MyBatis-Plus -->
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>最新版本</version>
</dependency>
<!-- ClickHouse -->
<dependency>
    <groupId>com.clickhouse</groupId>
    <artifactId>clickhouse-jdbc</artifactId>
    <version>最新版本</version>
</dependency>

配置 application.properties 或 application.yml 文件：




# MyBatis-Plus 配置
spring.datasource.url=jdbc:mysql://localhost:3306/yourdb
spring.datasource.username=yourusername
spring.datasource.password=yourpassword
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
 
# ClickHouse 配置
clickhouse.db.url=jdbc:clickhouse://localhost:8123
clickhouse.db.user=default
clickhouse.db.password=

创建 ClickHouse 的配置类：




import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.clickhouse.jdbc.ClickHouseDataSource;
 
@Configuration
public class ClickHouseConfig {
 
    @Value("${clickhouse.db.url}")
    private String clickHouseUrl;
 
    @Value("${clickhouse.db.user}")
    private String clickHouseUser;
 
    @Value("${clickhouse.db.password}")
    private String clickHousePassword;
 
    @Bean(name = "clickHouseDataSource")
    public ClickHouseDataSource clickHouseDataSource() {
        ClickHouseDataSource dataSource = new ClickHouseDataSource();
        dataSource.setUrl(clickHouseUrl);
        dataSource.setUser(clickHouseUser);
        dataSource.setPassword(clickHousePassword);
        return dataSource;
    }
}

创建 MyBatis-Plus 的 Mapper 接口和对应的 XML 文件：




import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.apache.ibatis.annotations.Mapper;
import org.springframework.stereotype.Repository;
 
@Mapper
@Repository
public interface YourEntityMapper extends BaseMapper<YourEntity> {
    // 自定义查询方法
}

创建实体类和对应的服务层：




import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
@Service
public class YourService {
 
    @Autowired
    private YourEntityMapper yourEntityMapper;
 
    // 使用 MyBatis-Plus 提供的方法或自定义方法
}

在 application.java 中启用 MyBatis-Plus 和 ClickHouse 配置：

2024-09-05

所有,java

在Spring Boot项目中集成ClickHouse数据库，并使用MyBatis Plus操作ClickHouse，可以通过以下步骤实现：

在pom.xml中添加ClickHouse的依赖：




<dependency>
    <groupId>com.clickhouse.jdbc</groupId>
    <artifactId>clickhouse-jdbc</artifactId>
    <version>版本号</version>
</dependency>

配置application.properties或application.yml文件，添加ClickHouse的数据源配置：




# ClickHouse 数据源配置
spring.datasource.clickhouse.driver-class-name=com.clickhouse.jdbc.ClickHouseDriver
spring.datasource.clickhouse.url=jdbc:clickhouse://localhost:8123/数据库名
spring.datasource.clickhouse.username=用户名
spring.datasource.clickhouse.password=密码

创建配置类，配置MyBatis Plus的数据源：




@Configuration
public class ClickHouseConfig {
 
    @Bean(name = "clickHouseDataSource")
    @ConfigurationProperties(prefix = "spring.datasource.clickhouse")
    public DataSource clickHouseDataSource() {
        return DataSourceBuilder.create().build();
    }
 
    @Bean(name = "clickHouseSqlSessionFactory")
    public SqlSessionFactory sqlSessionFactory(@Qualifier("clickHouseDataSource") DataSource clickHouseDataSource) throws Exception {
        SqlSessionFactoryBean bean = new SqlSessionFactoryBean();
        bean.setDataSource(clickHouseDataSource);
        return bean.getObject();
    }
 
    @Bean(name = "clickHouseSqlSessionTemplate")
    public SqlSessionTemplate sqlSessionTemplate(@Qualifier("clickHouseSqlSessionFactory") SqlSessionFactory sqlSessionFactory) {
        return new SqlSessionTemplate(sqlSessionFactory);
    }
}

使用MyBatis Plus提供的Mapper操作ClickHouse数据库：




public interface YourEntityMapper extends BaseMapper<YourEntity> {
    // 自定义的数据库操作方法
}

在Service层注入Mapper，进行数据库操作：




@Service
public class YourService {
 
    @Autowired
    private YourEntityMapper yourEntityMapper;
 
    public List<YourEntity> getYourEntityList() {
        return yourEntityMapper.selectList(null);
    }
}

以上步骤可以实现Spring Boot项目中集成ClickHouse数据库，并通过MyBatis Plus操作ClickHouse。如果需要集成多数据源，可以通过Spring的@Primary注解指定主数据源，并为每个数据源创建独立的配置类和SqlSessionFactoryBean实例。