ClickHouse_ClickHouse跨集群迁移数据表工具

发表于 | 分类于

ClickHouse_ClickHouse跨集群迁移数据表工具

关键命令

1
insert into sz.app_jdr_traffic_sz_all_chan_mvp_i_d_d_d select * from remote('11.102.245.98:9600',sz,app_jdr_traffic_sz_all_chan_mvp_i_d_d,'rw_sz_merchant_flow_slave_v1','RkDtzFEgMbxKmCH2MbkO') where partition >= '2023-12-11' and partition <= '2023-12-11' settings receive_timeout = 6000, send_timeout = 6000, insert_deduplicate = 0;

主要使用上述语句可以实现跨集群数据迁移，需要注意：

  1. 一般是读本地表写分布式表，因为跨集群的分片数和副本数不一定一致，需要指定新表的分片键，然后写分布式表，让分布式表帮助分片；
  2. 需要轮询源表的不同节点，分别读数，因为需要尽量分散读数操作，避免超出时间、内存等配额限制；
  3. 需要轮询源表的不同分区，分别读数，因为需要尽量分散读数操作，避免超出时间、内存等配额限制。

完整工具代码

启动命令

/usr/bin/python3 copyData.py –src_db eco_data_bp –src_table adm_eco_s10_afs_sku_sum_test_local –dest_db eco_data_bp –dest_table adm_eco_s10_afs_sku_sum_test_local

conf.json

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
{
    "srcCluster":{
        "host": "ckpub76.olap.jd.com",
        "port": 2000,
        "user": "default",
        "password": "***",
        "cluster_name": "LF10_CK_Pub_76",
        "tcp_http_mappings":
        {
            "9000": "8123",
            "9600": "8623",
            "9700": "8723",
            "9800": "8823",
            "9900": "8923"
        }
    },
    "destCluster":{
        "host": "ckpub163.olap.jd.com",
        "port": 2000,
        "user": "default",
        "password": "***",
        "cluster_name": "LFRH_CK_Pub_163",
        "tcp_http_mappings":
        {
            "9000": "8123",
            "9600": "8623",
            "9700": "8723",
            "9800": "8823",
            "9900": "8923"
        }
    },
    "taskSettings":
    {
        "sleep_second": 5,
        "parallelism": 15,
        "allow_to_drop_dest_partition": false,
        "check_count": true
    }
}

cluster.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
from multiprocessing import Pool
import requests
import time
import functools
import threading


class CantRetry(RuntimeError):
    ...


class ClickHouseException(RuntimeError):
    ...


def retry(times, sec):

    def wrapper(func):

        def inner_wrapper(*args, **kwargs):
            i = 0
            while True:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    i += 1
                    if i > times:
                        raise (e)
                    time.sleep(sec)
        return inner_wrapper

    return wrapper


@retry(2, 3)
def query_http(host, port, user, password, sql):
    ckurl = f'http://{user}:{password}@{host}:{port}/'
    response_result = requests.post(ckurl, data=sql.encode('utf-8'))
    if response_result.status_code != 200:
        raise ClickHouseException(
            f"Query executed failed:{response_result.text}")
    return response_result.text


class ClickHouseInstance:
    def __init__(self, host_address, tcp_port, http_port, cluster, shard, replica_num):
        self.host_address = host_address
        self.tcp_port = tcp_port
        self.http_port = http_port
        self.replica_num = replica_num
        self.shard = shard
        self.cluster = cluster
        self.user = cluster.user
        self.password = cluster.password

    def query_http(self, sql):
        return query_http(self.host_address, self.http_port, self.user, self.password, sql)


class Shard:
    def __init__(self, cluster, shard_num):
        self.shard_num = shard_num
        self.replicas = list()
        self.query_idx = 0
        self.cluster = cluster

    def append_replica(self, replica_num, host_address, tcp_port, http_port):
        self.replicas.append(ClickHouseInstance(
            host_address, tcp_port, http_port, self.cluster, self, replica_num))

    def query_all_replicas(self, sql):
        for replica in self.replicas:
            replica.query_http(sql)

    def query(self, sql):              
        idx = 0
        while idx < len(self.replicas): 
            try:    
                idx += 1
                self.replicas[idx].query_http(sql)
            except Exception as e:
                time.sleep(1)




# 一个ck实例，包含所有分片，以及defualt账户密码
class Cluster:
    def __init__(self, cluster_name, host, port, user, password, tcp_http_mappings):
        self.cluster_name = cluster_name
        self.domain_name = host
        self.port = port
        self.user = user
        self.password = password
        self.tcp_http_mappings = tcp_http_mappings
        self.shards = self.get_shards()

    def query(self, sql):
        return query_http(self.domain_name, self.port, self.user, self.password, sql)

    def query_one_replica(self, sql):
        return query_http(self.domain_name, self.port, self.user, self.password, sql)

    def query_all(self, sql):
        with Pool(len(self.shards)) as p:
            clusters = p.map(functools.partial(
                Shard.query_all_replicas, sql=sql), self.shards.values())

    def get_shards(self):
        shards = dict()
        get_shard_sql = f'''select shard_num, replica_num, host_address, port from system.clusters
        where cluster='{self.cluster_name}' order by shard_num, replica_num;'''
        result = self.query(get_shard_sql)

        for line in result.splitlines():
            shard_num, replica_num, host_address, tcp_port = line.split()
            shard = shards.get(shard_num, Shard(self, shard_num))
            shard.append_replica(replica_num, host_address,
                                 tcp_port, self.tcp_http_mappings[tcp_port])
            shards[shard_num] = shard
        return shards

copyData.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
#!/usr/bin/python3
#!/usr/bin/env python3

import codecs
import datetime
import argparse
from multiprocessing import Pool
from urllib import response
import os
import logging
import json
import sys
from cluster import *
import concurrent.futures
import random


FORMAT = '%(asctime)-15s [%(levelname)s] [%(filename)s:%(lineno)d] %(message)s'
logging.basicConfig(format=FORMAT)
logging.getLogger().setLevel(logging.INFO)

# =============================
# 数据拷贝task类的定义以及初始化
# =============================

class Conf:
    def __init__(self, config_path: str):
        try:
            with open(config_path, 'r') as load_f:
                self.loaded_json = json.load(load_f)
        except Exception as e:
            raise Exception(
                f"can't get conf fron file '{config_path}': {e}")


def get_cluster_by_config(config):
    return Cluster(config['cluster_name'], config['host'], config['port'],
                   config['user'], config['password'])


def get_partitions(cluster, database, table, range):
    get_paritions_count_sql = f'''select countDistinct(partition) from cluster(system_cluster, system.parts) where database='{database}' and table='{table}' ; '''
    logging.info('test2')
    paritions_count = cluster.query(get_paritions_count_sql).splitlines()[0]
    conditions = ''
    if range[0] is not None:
        conditions += f"and partition >= '{range[0]}'"
    if range[1] is not None:
        conditions += f"and partition <= '{range[1]}'"
        
    get_partitions_sql = f"select distinct(partition) from cluster(system_cluster, system.parts) where database='{database}' and table='{table}' {conditions} order by partition; "
    paritions = cluster.query(get_partitions_sql).splitlines()
    
    logging.info(f"get table {database}.{table} {len(paritions)}/{paritions_count} partitions by range {range} from cluster {cluster.cluster_name} ")
    return paritions


def get_condition(partition_key, partition):
    if partition == 'tuple()' or partition_key == '':
        return ''
    partition_split = partition.encode('utf-8').decode(
    'unicode_escape').replace('(', '').replace(')', '').split(',')

    if partition_key[0] == '(':
        partition_key = partition_key.replace('(', '').replace(')', '')
    partition_keys = partition_key.split(', ')
    if len(partition_keys) == 1:
        return 'where ' + f"toString({partition_keys[0]}) = '{partition_split[0]}'"
    conditions = [
        f"{a} = {b}" for a, b in zip(partition_keys, partition_split)
    ]
    return 'where ' + ' and '.join(conditions) 

class Task:
    def __init__(self, configs):
        self.src_cluster = Cluster(**configs["srcCluster"])
        self.dest_cluster = Cluster(**configs["destCluster"])

        self.sleep_second = configs['taskSettings'].get('sleep_second', 1)
        self.parallelism = configs['taskSettings'].get('parallelism', len(self.dest_cluster.shards))
        self.allow_to_drop_dest_partition = configs['taskSettings'].get('allow_to_drop_dest_partition', 0)

        self.src_db = None
        self.src_table = None
        self.dest_db = None
        self.dest_table = None
        self.dest_dis_table = None
        
        self.engine = None       
        self.is_dimension_table = False
        self.partition_key = None
        self.partitions = None
        self.shard_key = None
        

    def initialize_src(self, src_db, src_table, range):
        self.src_db = src_db
        self.src_table = src_table
        
        sql = f"select partition_key from system.tables where name = '{src_table}' and database = '{src_db}'"

        self.partition_key = self.src_cluster.query(sql).splitlines()[0]
        logging.info("get partition_key: %s", self.partition_key)
        
        self.partitions = get_partitions(self.src_cluster, src_db,
                                         src_table, range)
        
        sql = f"select count() from system.tables where name = '{src_table}' and database = '{src_db}' and create_table_query like '%replica_dict%'"
        
        self.is_dimension_table = (int(self.src_cluster.query(sql).splitlines()[0]) == 1)

        return

    def check_table_exists(self):
        sql = f"select count() from system.tables where name = '{self.dest_table}' and database = '{self.dest_db}'"
        try:
            result = self.dest_cluster.query(sql)
        except Exception as e:
            logging.error("check table %s.%s exists on dest cluster %s failed: %s", self.dest_db, self.dest_table, self.dest_cluster.cluster_name, e)
            sys.exit(-1)
            
        return int(result.splitlines()[0]) == 1

    def get_create_table_sql(self):
        sql = f"show create {self.src_db}.{self.src_table}"
        try:
            result = self.src_cluster.query(sql)
        except Exception as e:
            logging.error("get create table %s.%s sql %s failed: %s", self.src_db, self.src_table, sql, e)
            sys.exit(-1)
        
        return codecs.escape_decode(bytes(result,"utf-8"))[0].decode("utf-8")
    
    def create_dest_table(self):
        if(self.check_table_exists()):
            return
        else:
            logging.info('try to create table %s.%s on dest cluster', self.dest_db, self.dest_table) 
        
        create_table_sql = self.get_create_table_sql()
        
        create_table_sql = create_table_sql.replace(self.src_cluster.cluster_name, self.dest_cluster.cluster_name).replace(f"{self.src_db}.{self.src_table}", f"{self.dest_db}.{self.dest_table}")

        try:
            self.dest_cluster.query_all(create_table_sql)
        except Exception as e:
            logging.error("create table %s.%s sql %s failed: %s", self.dest_db, self.dest_table, create_table_sql, e)
            sys.exit(-1)
    
        
        
    def create_dest_dis_table(self):
        create_table_sql = f'''create table {self.dest_db}.`{self.dest_dis_table}` 
        as  {self.dest_db}.`{self.dest_table}`
        ENGINE = Distributed('{self.dest_cluster.cluster_name}', '{self.dest_db}', '{self.dest_table}', {self.shard_key}) '''
        try:
            self.dest_cluster.query_all(create_table_sql)
        except Exception as e:
            logging.error("create table %s.%s sql %s failed: %s", self.dest_db, self.dest_table, create_table_sql, e)
            sys.exit(-1)
 

    def initialize_dest(self, dest_db, dest_table, shard_key):
        self.shard_key = shard_key
        
        if dest_db is None:
            self.dest_db = self.src_table
        else:
            self.dest_db = dest_db

        if dest_table is None:           
            self.dest_table = self.src_table
        else:
            self.dest_table = dest_table
        
        self.dest_dis_table = f'{self.dest_table}_{int(time.time())}_dis'
            
        self.create_dest_table()   
        self.create_dest_dis_table() 

    def check_partition(self, condition, partition):

        try:
            source_func = f'''cluster( {self.src_cluster.cluster_name}, {self.src_db}.`{self.src_table}`)'''   
            desc_func = f'''cluster( {self.dest_cluster.cluster_name}, {self.dest_db}.`{self.dest_table}`)'''
            
            if self.engine == 'ReplicatedReplacingMergeTree':
                source_func += ''' final '''
                desc_func += ''' final '''
            
                  
            src_sql = f"select count() from {source_func}  {condition} settings do_not_merge_across_partitions_select_final = 1"
            dest_sql = f"select count() from {desc_func}  {condition} settings do_not_merge_across_partitions_select_final = 1"
            
            src_cnt = int(self.src_cluster.query(src_sql).splitlines()[0])
            dest_cnt = int(self.dest_cluster.query(dest_sql).splitlines()[0])        
            
                 
            if src_cnt != dest_cnt:
                logging.info(
                    f'{self.dest_cluster.cluster_name}中表 {self.dest_db}.`{self.dest_table}` 分区 {partition} 数据不一致, 源集群: {src_cnt}, 目标集群: {dest_cnt}, 尝试重新写入'
                )
                if dest_cnt != 0:
                    if self.allow_to_drop_dest_partition == False:
                        logging.error( f'{self.dest_cluster.cluster_name} 中表 {self.dest_db}.`{self.dest_table}` 目标集群分区 {partition} 数据不为空，不允许删除，跳过分区 {partition}')
                        return True
                    
                    logging.info(
                        f'{self.dest_cluster.cluster_name} 中表 {self.dest_db}.`{self.dest_table}` 目标集群分区 {partition} 数据不为空: {dest_cnt}, 尝试删除'
                    )
                    
                    drop_partition_sql = f"alter table {self.dest_db}.`{self.dest_table}`  on cluster {self.dest_cluster.cluster_name} drop partition '{partition}'"
                    if partition[0] == '(':
                        new_partition_str = partition.replace('\\', '')                     
                        drop_partition_sql = f"alter table {self.dest_db}.`{self.dest_table}`  on cluster {self.dest_cluster.cluster_name} drop partition {new_partition_str}"                
                    self.dest_cluster.query(drop_partition_sql)
                    time.sleep(1)
                return False
            else:
                logging.info(f'{self.dest_cluster.cluster_name}中表 {self.dest_db}.`{self.dest_table}` 分区 {partition} 数据一致, 源集群: {src_cnt}, 目标集群: {dest_cnt}')
        except Exception as e:
            logging.error(e)
            logging.error(f"验证分区 {partition} 出错，请重试")
            raise(Exception)
        return True

    def insert_data(self, src_shard, condition):
        idx = random.randint(0, len(src_shard.replicas)-1)
        replica = src_shard.replicas[idx]
                
        source_func = f'''remote('{replica.host_address}:{replica.tcp_port}',{self.src_db},{self.src_table},'{self.src_cluster.user}','{self.src_cluster.password}')'''
            
        settings = '''settings receive_timeout = 6000, send_timeout = 6000, insert_deduplicate = 0'''
        
        if self.engine == 'ReplicatedReplacingMergeTree':
            settings += ''', do_not_merge_across_partitions_select_final = 1'''
            source_func += ''' final '''
    
        sql = f'''insert into {self.dest_db}.`{self.dest_dis_table}`
        select * from {source_func} {condition} 
        {settings};'''
        
        # logging.info(sql)
        self.dest_cluster.query(sql)

    @retry(2, 3)
    def copy_parition(self, partition, condition):
        # 验证，如果有问题删除
        if self.check_partition(condition, partition):
            return
        logging.info(f"准备拷贝分区 {partition}")
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.parallelism) as executor:
            futures = list()               
            for shard_num, shard in self.src_cluster.shards.items():
                logging.info(f"准备拷贝分片 {shard_num}")
                futures.append(executor.submit(Task.insert_data, self, shard, condition))
            concurrent.futures.wait(futures)
            for i in range(len(futures)):                   
                exception = futures[i].exception()
                # handle exceptional case
                if exception:
                    logging.error(f"源集群 {self.src_cluster.cluster_name} 分片 {i+1} 分区 {partition} 拷贝失败: {exception}")
                    raise(exception)
          

    def run(self):
        for partition in self.partitions:
            condition = get_condition(self.partition_key, partition)           
            self.copy_parition(partition, condition)
        
    def drop_dest_dis_table(self):
        drop_table_sql = f"drop table {self.dest_db}.`{self.dest_dis_table}`"
        self.dest_cluster.query_all(drop_table_sql)
         


def copy_data(args):
    logging.info(args)

    logging.info(f"解析配置文件")
    config = Conf(args.config)
    
    logging.info(f"初始化任务")
    copy_task = Task(config.loaded_json) 
    
    logging.info(f"从源表{args.src_db}中获取迁移数据信息")
    copy_task.initialize_src(args.src_db, args.src_table, (args.min_partition, args.max_partition))
    
    copy_task.initialize_dest(args.dest_db, args.dest_table, args.shard_key)
    try:
        copy_task.run()
    except Exception as e:
        logging.info(f"迁移数据失败中止: {e}")
    finally:
         copy_task.drop_dest_dis_table()
    
   

# =============================
# 主函数
# =============================

if __name__ == "__main__":
    
    parser = argparse.ArgumentParser(description='ClickHouse集群数据迁移脚本')
    parser.add_argument('--src_db', type=str, required=True, help='源数据库')
    parser.add_argument('--src_table', type=str, required=True, help='源本地表')
    parser.add_argument('--dest_db', type=str, help='目标数据库')
    parser.add_argument('--dest_table', type=str, help='目标本地表')
    parser.add_argument('--min_partition', type=str, help='最小分区')
    parser.add_argument('--max_partition', type=str, help='最大分区')

    parser.add_argument('--config',
                        type=str,
                        default=f'conf.json',
                        help='配置文件')
    parser.add_argument('--shard_key',
                        type=str,
                        default='rand()',
                        help='分片之间分发的规则')
    args = parser.parse_args()
    copy_data(args)