MySQL 到 ClickHouse (进阶)

简述

本文简要介绍 CloudCanal 参考 ClickHouse 官方实现的 MaterializeMySQL 表引擎，将 ReplacingMergeTree 作为对端主力表引擎进行数据同步链路构建。

新方案优势包括

表不存在额外的字段
按 order by key 严格合并(单节点)
可以设置按时间间隔自动 optimize 表(autoOptimizeThresholdSec参数)
支持 DDL 同步

技术点

结构迁移

目前到 ClickHouse 的结构迁移中，默认选择 ReplacingMergeTree 作为表引擎，源主键作为 sortKey (无主键表则是 tuple)，如下示例:

 CREATE TABLE console.worker_stats
(
    `id` Int64,
    `gmt_create` DateTime,
    `worker_id` Int64,
    `cpu_stat` String,
    `mem_stat` String,
    `disk_stat` String
)
ENGINE = ReplacingMergeTree()
ORDER BY id
SETTINGS index_granularity = 8192

写数据

新方案全量仍然按照标准 batch 导入，增量和 CollapsingMergeTree 作为表引擎的区别在于

转换 Insert、Update 操作为 Insert
Delete 操作单独通过 alter table delete 语句进行操作

所以 Delete 操作如果较多，增量同步性能会急剧下降，建议 delete RPS 不超过 50。

  switch (rowChange.getEventType()) {
            case INSERT:
            case UPDATE: {
                for (CanalRowData rowData : rowChange.getRowDatasList()) {
                    CkTableBatchData.RecordWithState addRecord = new CkTableBatchData.RecordWithState(CanalEventType.INSERT, rowData.getAfterColumnsList());
                    batchData.getRecords().add(addRecord);
                }

                break;
            }
            case DELETE: {
                for (CanalRowData rowData : rowChange.getRowDatasList()) {
                    CkTableBatchData.RecordWithState delRecord = new CkTableBatchData.RecordWithState(CanalEventType.DELETE, rowData.getBeforePkColumnsList());
                    batchData.getRecords().add(delRecord);
                    batchData.setHasDelete(true);
                }
                break;
            }
            default:
                throw new CanalException("not supported event type,eventType:" + rowChange.getEventType());
 }