Posted by Lance Lee on Sunday, May 24, 2020






  • 连接到Zookeeper然后接收:
    • 复制任务。
    • 复制任务的状态。
  • 执行任务。 每个运行中的进程选择源集群中“最近”的分片,并将数据拷贝到目标集群中,如果有必要的话还会对数据重新分片。

clickhouse-copier 跟踪ZooKeeper中的更改并动态应用它们。


运行 Clickhouse-copier


$ clickhouse-copier copier --daemon --config zookeeper.xml --task-path /task/path --base-dir /path/to/dir


  • daemon — 在后台模式下启动clickhouse-copier。
  • config — ZooKeeper. xml文件的路径,其中包含ZooKeeper连接的参数。
  • task-path — ZooKeeper节点的路径。此节点用于同步clickhouse-copier进程和存储任务。任务存储在$task-path/description中。
  • task-file — 可选的任务配置的文件路径,用于初始上传到ZooKeeper。
  • task-upload-force — 强制上传任务文件,即使节点已经存在。
  • base-dir — 日志和附属文件的路径。当它启动时,clickhouse-copier在$base-dir中创建clickhouse-copier_YYYYMMHHSS_<PID>子目录。如果省略此参数,则在启动clickhouse-copier的目录中创建目录。

Zookeeper.xml 格式


        <node index="1">


    <!-- Configuration of clusters as in an ordinary server config -->


    <!-- How many simultaneously active workers are possible. If you run more workers superfluous workers will sleep. -->

    <!-- Setting used to fetch (pull) data from source cluster tables -->

    <!-- Setting used to insert (push) data to destination cluster tables -->

    <!-- Common setting for fetch (pull) and insert (push) operations. Also, copier process context uses it.
         They are overlaid by <settings_pull/> and <settings_push/> respectively. -->
        <!-- Sync insert is set forcibly, leave it here just in case. -->

    <!-- Copying tasks description.
         You could specify several table task in the same task description (in the same ZooKeeper node), they will be performed
        <!-- A table task, copies one table. -->
            <!-- Source cluster name (from <remote_servers/> section) and tables in it that should be copied -->

            <!-- Destination cluster name and tables in which the data should be inserted -->

            <!-- Engine of destination tables.
                 If destination tables have not be created, workers create them using columns definition from source tables and engine
                 definition from here.

                 NOTE: If the first worker starts insert data and detects that destination partition is not empty then the partition will
                 be dropped and refilled, take it into account if you already have some data in destination tables. You could directly
                 specify partitions that should be copied in <enabled_partitions/>, they should be in quoted format like partition column of
            ENGINE=ReplicatedMergeTree('/clickhouse/tables/{cluster}/{shard}/hits2', '{replica}')
            PARTITION BY toMonday(date)
            ORDER BY (CounterID, EventDate)

            <!-- Sharding key used to insert data to destination cluster -->
            <sharding_key>jumpConsistentHash(intHash64(UserID), 2)</sharding_key>

            <!-- Optional expression that filter data while pull them from source servers -->
            <where_condition>CounterID != 0</where_condition>

            <!-- This section specifies partitions that should be copied, other partition will be ignored.
                 Partition names should have the same format as
                 partition column of table (i.e. a quoted text).
                 Since partition key of source and destination cluster could be different,
                 these partition names specify destination partitions.

                 NOTE: In spite of this section is optional (if it is not specified, all partitions will be copied),
                 it is strictly recommended to specify them explicitly.
                 If you already have some ready partitions on destination cluster they
                 will be removed at the start of the copying since they will be interpeted
                 as unfinished data from the previous copying!!!

        <!-- Next table to copy. It is not copied until previous table is copying. -->



Attack On Programmer