复制的注意点

目前只有合并树系列的表 具有复制的功能。

复制的实现原理

clickhouse的复制 是基于 zookeeper来实现的。zookeeper 在clickhouse 复制的实现之中 扮演了 元数据存储、日志框架、分布式协调服务 三重角色。
每个clickhouse 节点都会监视 zookeeper /clickhouse/tables/分片号/数据库名/表名/log 节点下的信息。当向一个节点的可复制表写入数据时，节点会向 /clickhouse/tables/分片号/数据库名/表名/log 目录下添加任务节点。另一个副本节点监控到变化，执行fetch操作（在zookeeper之中的数据变化，这里忽略），从对应的节点拉取数据。

fetch

fetch 的定义

复制表的一个副本从另一个副本 通过http协议 克隆数据，这个过程被称为fetch。

fetch 的实现

每个clickhouse 节点 专门维护一个线程池 用以执行fetch操作。fetch的实质 是通过http协议下载数据。

决定 fetch 线程池大小的参数

background_fetches_pool_size

clickhouse 节点 暴露的用于执行fetch的端口是哪个

config.xml 文件之中的配置参数

 <interserver_http_port>9009</interserver_http_port>

通过http fetch的时候，接收方的超时时间

replicated_fetches_http_receive_timeout 参数决定，是合并树表级别的配置参数。

通过http fetch的时候，发送方的超时时间

replicated_fetches_http_send_timeout

每个节点的fetch 并发度限制

replicated_max_parallel_fetches_for_host

限制 fetch 收 发 消耗的网络带宽大小

max_replicated_sends_network_bandwidth_for_server： 服务器级别的设置（限制所有表的总量）
max_replicated_fetches_network_bandwidth_for_server：服务器级别的设置（限制所有表的总量）
max_replicated_fetches_network_bandwidth ： 表级别的设置
max_replicated_sends_network_bandwidth ： 表级别的设置

fetch 操作会占据的资源

带宽 cpu 内存。 主要需要注意的是带宽。

replica 在 zookeeper 的清理机制

• max_replicated_logs_to_keep：当一个副本节点挂掉的时候，zookeeper的log队列最多保持多少个任务节点。
• min_replicated_logs_to_keep：在log队列之中最少保持多少个任务节点。即使任务节点已经被执行，但是还是需要保持配置数量的节点数。
• cleanup_delay_period：合并树级别的配置参数，决定了后台执行复制表队列的清理周期。
  如果发现队列之中的节点数过多的话，可以稍稍调小一下参数，默认的取值为30s。

复制表下的合并

1：execute_merges_on_single_replica_time_threshold 参数决定副本开始合并的时间。假设有两个副本，一个副本开始合并，那么另一个副本是需要fetch 合并后的数据呢，还是自己也执行合并。所以就取决于参数 execute_merges_on_single_replica_time_threshold的值。取值为0，表示每个副本自己触发合并。当取值为某一个整数时，副本A开始执行合并，副本B等待一段时间，如果副本A的合并执行完成，那么副本B通过fetch获取数据，否则副本B自己本地开始执行合并。
2：当log entry的创建时间 超过 prefer_fetch_merged_part_time_threshold，
并且执行合并的数据量的大小 超过 prefer_fetch_merged_part_size_threshold,
那么执行fetch 来加速合并。

可复制表的写入

复制的相关配置参数

复制表下的合并参数

• execute_merges_on_single_replica_time_threshold：默认的取值为0，表示各个节点在自己本地进行合并。含义是，使某个副本首先开始合并，如果在指定的时间内完成合并，通过fetch来获得合并后的数据。作用：用来缓解合并对cpu造成的压力
• prefer_fetch_merged_part_time_threshold：当log entry的创建时间 超过 prefer_fetch_merged_part_time_threshold，并且执行合并的数据量的大小 超过 prefer_fetch_merged_part_size_threshold,那么执行fetch 来加速合并

复制表下的写入参数

• insert_quorum
• insert_quorum_timeout
• insert_quorum_parallel

复制表下的查询参数

• max_replica_delay_for_distributed_queries：当执行查询的时候，会选择某个副本进行数据查询，这是一个选择的标志，落后多久的副本仍然可以作为查询的数据来源。
• fallback_to_stale_replicas_for_distributed_queries：当更新的副本不可用的时候，强制使用过时的副本进行查询，保证查询可行
• select_sequential_consistency：一致性查询，只有insert_quorum_parallel被禁用的时候，才会生效。

复制表相关的系统表

system.replicas
system.replicated_fetches
system.replication_queue

复制的监控

各节点配置的 fetch 任务队列长度

select 
   hostName() as hostName,
   value 
from clusterAllReplicas('default', 'system.settings')
where name = 'background_fetches_pool_size'
order by hostName;

各节点正在执行的fetch任务数

select 
  hostName,
  count(*) as num 
from (
select 
   hostName() as hostName,
   database,
   table
from
  clusterAllReplicas('集群名', 'system.replicated_fetches')
) group by 
hostName

各节点正在执行的fetch任务的信息，根据耗时时间进行排序

select 
   hostName() as hostName,
   result_part_name,
   source_replica_path,
   source_replica_hostname,
   source_replica_port,
   database,
   table,
   elapsed,
   progress,
   total_size_bytes_compressed, --fetch所要读取的所有的数据量 
   bytes_read_compressed, -- 已经读取的数据量
   thread_id,
   to_detached
from 
  clusterAllReplicas('default', 'system.replicated_fetches')
order by elapsed desc 

节点后台等待执行的fetch 任务数

select 
   hostName() as hostName,
   count(*) as num 
from 
   clusterAllReplicas('default', 'system.replication_queue') 
where 
   type = 'GET_PART'
 and is_currently_executing = 0 
order by hostName 

节点后台等待执行的fetch 任务详细信息，根据等待时间倒序排序

select 
   hostName() as hostName,
   toUnixTimestamp(now()) - toUnixTimestamp(create_time) as task_wait_time,
   source_replica, --来源副本名
   replica_name,  -- 目标副本名
   position,      --任务位置 
   node_name,     --任务节点名
   database,      -- 数据库名
   table,         -- 表名
   new_part_name, -- 生成的副本的名字
   create_time,   --任务创建时间
   num_postponed, -- 推迟次数
   postpone_reason, -- 推迟原因
   last_postpone_time --上次推迟时间
from 
   clusterAllReplicas('default', 'system.replication_queue') 
where 
   type = 'GET_PART'
 and is_currently_executing = 0 
order by task_wait_time desc 

重试次数大于 指定限制的 fetch 任务数

select 
  hostName,
  count(*) as num
from (
select 
  hostName() as hostName,
  replica_name
from 
   clusterAllReplicas('集群名', 'system.replication_queue') 
where 
   num_tries > 设置的重试次数
   and  type = 'GET_PART'
)
where hostName in (${hostName})
group by
   hostName 

重试次数大于 指定限制的 fetch 任务的详情

select 
   *
from (
select 
   hostName() as hostName,
   num_tries,
   last_exception,
   last_attempt_time,
   source_replica,
   replica_name,
   database,
   table,
   new_part_name,
   node_name,
   position,
   create_time
from 
  clusterAllReplicas('集群名', 'system.replication_queue') 
where 
   num_tries > 重试次数
   and  type = 'GET_PART'
)
where 
order by num_tries,create_time desc 
limit 30

后台part fetch 失败的次数

select * from system,events where event = 'ReplicatedPartFailedFetches'

查看fetch 的繁忙程度

通过查看任务还未执行的延时时间，如果出现有的任务很久之前已经提交了，但是很久未被执行，可以来反应

select 
  hostName() as hostName,
  toUnixTimestamp(now()) - toUnixTimestamp(create_time) as task_delay,
   * 
from 
  clusterAllReplicas('集群名', ' system.replication_queue ')
where is_currently_executing = 0
   and type = 'GET_PART'
order by 
   task_delay desc  limit 10