clickhouse集群迁移实践

原文链接

背景

现有的ck集群没有副本，磁盘是12块盘的stat盘，存在磁盘故障导致数据丢失的风险，而依赖zk的双副本策略又由于zk性能存在瓶颈影响集群的可用性，故而使用带三副本的高效云盘替代stat盘，规避数据丢失的风险。

当前ck的写入程序使用的是统一的域名，由域名查询到对应的ip节点来建立tcp连接。对于ck的查询，使用的是内部的一个代理，该代理配置了集群的ip，由代理去轮询ip进行查询。

在数据迁移的过程中，需要保证集群写入和查询都不受影响，关键在于控制好查询和写入的节点。

迁移方案

1. 新节点安装ck，同步元数据，元数据管理覆盖新节点ip

新节点安装时，配置文件的remote_servers中的集群配置和旧集群一致，新节点只是做一个查询转发。

从旧节点中导出非system表的建表语句，由于一份表对应着一个本地表和一个分布式表，因此需要先创建完本地表。

讯享网

导出本地表 echo "select create_table_query||';' from system.tables where database != 'system' and engine!='Distributed' order by name desc" | /data/clickhouse/bin/clickhouse-client --password xxx --port 9000 > localtable.txt 导出分布式表的ddl echo "select create_table_query||';' from system.tables where database != 'system' and engine='Distributed' order by name desc" | /data/clickhouse/bin/clickhouse-client --password xxx --port 9000 > distable.txt 

将建表语句发送到新节点执行

讯享网执行本地表的ddl ./sendfile.sh ckip.txt localtable.txt distable.txt ./runcmd.sh ckip.txt "/data/clickhouse/bin/clickhouse-client --password xxx -mn < /home/jinyaqia/ck_tool/localtable.txt" 执行分布式表的ddl ./runcmd.sh ckip.txt "/data/clickhouse/bin/clickhouse-client --password xxx -mn < /home/jinyaqia/ck_tool/distable.txt" 

其中

# sendfile.sh # 用于发送文件 #!/usr/bin/env bash file='' array="$@" i=0 for el in $array do if [ ${i} -ne 0 ] then file=$file' '${el} fi let i++ done for ip in $(cat $1|grep -v "^#.*\|^$") do cmd="scp ${file} $ip:~/ck_tool/" $cmd & echo $ip,$cmd done wait echo 'done' 

讯享网# runcmd.sh # 用于执行命令 #!/usr/bin/env bash for ip in $(cat $1|grep -v "^#.*\|^$") do echo $ip ssh $ip "${2}" & echo "end ${ip}" done wait 

2. 配置查询代理，从新节点查询数据

3. 滚动迁移

3.1 先停掉待迁移节点的写入

域名dns管理去掉待迁移节点，这里需要写入端没有直接缓存ck节点ip，确保在数据复制的过程中，新的数据不会写入待迁移的节点

3.2 ck节点一对一复制数据

采用clickhouse copier的方式进行迁移，生成所有本地表的迁移配置，调用迁移脚本进行数据复制

1. 导出需要进行迁移的表，可以从system.tables中查询并导出

echo "select database||'.'||name,engine_full from system.tables where database != 'system' and engine not in ('Distributed','View','MaterializedView') order by name desc" | /data/clickhouse/bin/clickhouse-client --password xxx --port 9000 > dbtable.txt.all 

2. 将脚本和文件都上传到接口机，由接口机发送到新节点,脚本内容后文给出

讯享网dbtable.txt.all 需要迁移的表 dbtable.txt.todo 需要执行迁移的表 table_copy_copier.py 迁移脚本 table_check.py 数据一致性检查脚本 get_error.py 处理迁移异常的表 ipmap.txt 旧节点复制到新节点的ip对应关系 copier_template.xml 复制工具配置模板 zk.xml 复制工具zk配置 发送命令 ./sendfile.sh ckip.txt dbtable.txt.all dbtable.txt.todo table_copy_copier.py table_check.py get_error.py copier_template.xml zk.xml 

3. 执行复制

nohup python3 table_copy_copier.py 2>&1 >copy.log & 

4. 校对、检查和重跑

讯享网批量校对 python3 table_check.py python3 get_error.py 有异常的表会写入dbtable.txt.1 其中新节点数据条数少的写入error.txt.less 新节点数据条数多了的写入error.txt.more 手动核对下dbtable.txt.1里的表迁移异常的原因，可以查clickhouse-copier的日志，如果要重跑，则生成新的dbtable.txt.todo, 删掉zk中对应表的节点，然后调用table_copy_copier.py重新复制 

3.3 更新ck集群配置remote_servers，增加新节点，去掉旧节点，等待ck server自动加载更新配置，此时查询会落到新节点上

3.4 配置域名，使数据写入新节点

4. 退回机器

注意点

1. 该方案在数据复制的时候要求旧节点和新节点都不会写数据，这个需要写入的时候不写分布式表
2. 复制任务重跑时，需要删掉zk节点的数据
3. 由于复制任务会在zk上创建比较多的节点数，为了降低zk的延迟，不要用ck集群的zk，可以单独搭建一个zk集群
4. table_copy_copier.py调用 clickhouse copier命令后，即使主进程关闭了，copier任务也会一直运行
5. table_copy_copier.py配置了60个子线程用于调用copier任务，如果要降低带宽的使用，可以调低并行度

附录：

table_copy_copier.py

import sys,json; import socket; import os; from concurrent.futures import ThreadPoolExecutor,as_completed import queue,threading; import time,functools def get_time(): time_now=int(time.time()) time_local=time.localtime(time_now) return time.strftime("%Y-%m-%d %H:%M:%S",time_local) def exec_command(fo,db,table,command_template,ip,task_file): try: task_path='/copytask/{}/{}/{}'.format(ip,db,table) base_dir=table if not os.path.exists(base_dir): os.makedirs(base_dir) command=command_template.format(task_path=task_path,base_dir=base_dir,task_file=task_file) print("start thread num={} command={} pid={}".format(threading.get_ident(),command,os.getpid())) f = os.popen(command) print("done thread num={} command={} pid={}".format(threading.get_ident(),command,os.getpid())) res="start {}.{} time={}\n".format(db,table,get_time()) fo.write(res) fo.flush() except: error="start {}.{} error\n".format(db,table) fo.write(error) fo.flush