共计 3578 个字符,预计需要花费 9 分钟才能阅读完成。
简述
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 id
SETTINGS index_granularity = 8192
ClickHouse 表引擎中,CollapsingMergeTree 和 VersionedCollapsingMergeTree 都能通过标记位按规定折叠数据,从而达到更新和删除的成果。VersionedCollapsingMergeTree 相比 CollapsingMergeTree 劣势在于同一条数据的不同变更能够乱序写入,然而 CloudCanal 抉择 CollapsingMergeTree 次要起因在于 2 点
-
- CloudCanal 中同一条记录必然是按源库变更程序写入,不存在乱序状况
-
- 不须要保护 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;
}
- 并行写入
将按表归组的数据应用并行执行框架执行,具体不详述。
举个 ” 栗子 ”
- 增加数据源
- 创立工作,抉择数据源和库,并连贯胜利,点击下一步
- 抉择数据同步,倡议规格至多抉择 1 GB. 目前 MySQL->ClickHouse 构造迁徙主动过滤,所以抉择有效。点击下一步
- 抉择表,默认 ClickHouse 上创立
CollapsingMergeTree
表引擎,并主动增加__cc_ck_sign
折叠标记字段。点击下一步 - 抉择字段,点击下一步
- 创立工作
- 期待工作主动构造迁徙、全量迁徙、数据同步追上
- 造点 Insert、Update、Delete 负载
- 提早追平状态,进行负载
- 查看源端 MySQL 表数据,以其中一张表为例
- 查看对端 ClickHouse 表数据,不统一?!!
- 手动优化下表,数据统一。尽管能够期待 ClickHouse 主动优化,然而如果须要间接失去精确后果,可手动优化(留神: 手动优化可能导致数据库机器压力过大)
常见问题
我在 ClickHouse 上曾经创立了表怎么办?
目前比拟倡议间接应用 CloudCanal 主动构造迁徙的形式来创立工作。
如果已建表为 CollapsingMergeTree 表引擎,请将标记位字段改成 __cc_ck_sign
Int8 DEFAULT 1`,再创立工作(此时就不再主动构造迁徙,而是应用已存在表)。
如果为其余表引擎,临时不反对(次要是不反对增量能力,须要 CloudCanal 进一步摸索)。
同步过来的数据什么时候合并?
当 CloudCanal 同步数据到 ClickHouse 时,ClickHouse 并不会实时合并数据,也没有一致性可言,所以个别状况是期待合并,或者间接手动合并 (造成机器高负载、高 IO), 如 optimize table worker_stats FINAL
。
DDL 怎么做?
目前 CloudCanal 还未反对到 ClickHouse 的 DDL 同步,产品实现上,目前是疏忽的。所以如果做 DDL,加字段倡议对端先加,再加源端,减字段反之。
总结
本文简要介绍了 CloudCanal 实现 MySQL(RDS)到 ClickHouse 数据迁徙同步的能力,具备一站式、数据实时特点,从技术点、例子、以及常见问题角度开展。文章如有谬误,烦请大家勘误,后续也欢送大家试用,提供贵重的意见和倡议。
CloudCanal- 收费好用的企业级数据同步工具,欢送品鉴。
理解产品能够查看官方网站:http://www.clougence.com
CloudCanal 社区:https://www.askcug.com/