2021年6月29日星期二

MySQL到ClickHouse实时同步-CloudCanal实战

简述

CloudCanal 近期实现了 MySQL（RDS） 到 ClickHouse 实时同步的能力，功能包含全量数据迁移、增量数据迁移、结构迁移能力，以及附带的监控、告警、HA等能力(平台自带)。

ClickHouse 本身并不直接支持 Update 和 Delete 能力，但是他自带的 MergeTree 系列表中 CollapsingMergeTree 和 VersionedCollapsingMergeTree 可变相实现实时增量的目的，并且性能完全够用,能够比较轻松达到 1k RPS 以上的能力。

接下来的文章，简要介绍 CloudCanal 是如何实现这个能力，以及作为用户我们怎么比较好的使用这个能力。

技术点

结构迁移

CloudCanal 默认提供结构迁移，默认选择 CollapsingMergeTree 作为表引擎，并增加一个默认字段 __cc_ck_sign，源主键作为 sortKey，如下示例:

 CREATE TABLE console.worker_stats( `id` Int64, `gmt_create` DateTime, `worker_id` Int64, `cpu_stat` String, `mem_stat` String, `disk_stat` String, `__cc_ck_sign` Int8 DEFAULT 1)ENGINE = CollapsingMergeTree(__cc_ck_sign)ORDER BY idSETTINGS index_granularity = 8192

ClickHouse 表引擎中，CollapsingMergeTree 和 VersionedCollapsingMergeTree 都能通过标记位按规则折叠数据，从而达到更新和删除的效果。VersionedCollapsingMergeTree 相比 CollapsingMergeTree 优势在于同一条数据的不同变更可以乱序写入，但是 CloudCanal 选择 CollapsingMergeTree 主要原因在于2点

    1. CloudCanal 中同一条记录必定是按源库变更顺序写入，不存在乱序情况
    1. 不需要维护 VersionedCollapsingMergeTree 中的 Version 字段(版本，也可以起其他名字)

所以 CloudCanal 选择了 CollapsingMergeTree 作为默认表引擎。

写数据

CloudCanal 写数据主要包含全量和增量两种，即单次搬迁存量数据和长期同步，两者写入略有不同。全量写入对端主要工作是批量和多线程，因为 CloudCanal 结构迁移默认设置了标记位字段 __cc_ck_sign default 值为 1, 所以就不需要做特殊处理。

对于增量, CloudCanal 则需要做 3 件事情。

  • 转换 Update、Delete 操作为 Insert
    这一步有两件事情要做，第一件是按照操作类型，填充标记字段值，其中 Insert 和 Update 为 1 ，Delete 为 -1 ，第二件是将对应增量数据的前镜像或者后镜像填充到结果记录中，以便后续 insert 写入。
 for (CanalRowChange rowChange : rowChanges) {   switch (rowChange.getEventType()) {    case INSERT: {     for (CanalRowData rowData : rowChange.getRowDatasList()) {      rowData.getAfterColumnsList().add(nonDeleteCol);      records.add(rowData.getAfterColumnsList());     }     break;    }    case UPDATE: {     for (CanalRowData rowData : rowChange.getRowDatasList()) {      rowData.getBeforeColumnsList().add(deleteCol);      records.add(rowData.getBeforeColumnsList());      rowData.getAfterColumnsList().add(nonDeleteCol);      records.add(rowData.getAfterColumnsList());     }     break;    }    case DELETE: {     for (CanalRowData rowData : rowChange.getRowDatasList()) {      rowData.getBeforeColumnsList().add(deleteCol);      records.add(rowData.getBeforeColumnsList());     }     break;    }    default:     throw new CanalException("not supported event type,eventType:" + rowChange.getEventType());   }  }
  • 按表归组
    因为 IUD 操作已全部转换为 Insert, 且为全镜像(所有字段都填充了值),所以可以按表归组，然后批量写入。即使单线程也能满足大部分场景的同步性能要求。
protected Map<TableUnit, List<CanalRowChange>> groupByTable(IncrementMessage message) {  Map<TableUnit, List<CanalRowChange>> data = new HashMap<>();  for (ParsedEntry entry : message.getEntries()) {   if (entry.getEntryType() == CanalEntryType.ROWDATA) {    CanalRowChange rowChange = entry.getRowChange();    if (!rowChange.isDdl()) {     List<CanalRowChange> changes = data.computeIfAbsent(new TableUnit(entry.getHeader().getSchemaName(), entry.getHeader().getTableName()), k -> new ArrayList<>());     changes.add(rowChange);    }   }  }  return data; }
  • 并行写入
    将按表归组的数据使用并行执行框架执行，具体不详述。

举个"栗子"

发帖者 时间：

没有评论:

发表评论

订阅： 博文评论 (Atom)